RAG_pdf_Ollama.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "eeaf36cf-969e-4632-b5eb-ac7bdc6e6df3",
   "metadata": {},
   "source": [
    "### RAG document search using OpenAI"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "4962e537-05af-4b08-adc8-87de5b5be7f1",
   "metadata": {},
   "outputs": [],
   "source": [
    "from PyPDF2 import PdfReader\n",
    "from langchain.embeddings.openai import OpenAIEmbeddings\n",
    "from langchain.text_splitter import CharacterTextSplitter\n",
    "from langchain.document_loaders import OnlinePDFLoader\n",
    "from langchain_community.embeddings import OllamaEmbeddings\n",
    "from langchain_community.vectorstores import Chroma"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "3c53cd8f-c8cb-4739-af99-6ea81776aa19",
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "os.environ['OPENAI_API_KEY'] = 'your_openai_key'\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "766c9980-85d6-4983-911d-bf87616d5125",
   "metadata": {},
   "outputs": [],
   "source": [
    "pdfreader = PdfReader(r\"C:\\Users\\ajayk\\Desktop\\OReilly Definitive Guide_delta_lake.pdf\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "f9437cda-b956-4b98-a505-44c41841fed9",
   "metadata": {},
   "outputs": [],
   "source": [
    "from typing_extensions import Concatenate\n",
    "# read text from pdf\n",
    "raw_text = ''\n",
    "for i, page in enumerate(pdfreader.pages):\n",
    "    content = page.extract_text()\n",
    "    if content:\n",
    "        raw_text += content"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "ae952d55-5930-437b-a451-419db3c5a6e0",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'With Early Release ebooks, you get books in their earliest\\nform—the authors’ raw and unedited content as they write—\\nso you can take advantage of these technologies long before\\nthe official release of these titles.\\nDenny Lee, Tathagata Das, and Vini JaiswalDelta Lake: The Definitive  Guide\\nModern Data Lakehouse Architectures\\nwith Delta Lake\\nBoston Farnham Sebastopol Tokyo Beijing Boston Farnham Sebastopol Tokyo Beijing978-1-098-10452-8Delta Lake: The Definitive  Guide\\nby Denny Lee, Tathagata Das, and Vini Jaiswal\\nCopyright © 2022 O’Reilly Media. All rights reserved.\\nPrinted in the United States of America.\\nPublished by O’Reilly Media, Inc., 1005 Gravenstein Highway North, Sebastopol, CA 95472.\\nO’Reilly books may be purchased for educational, business, or sales promotional use. Online editions are\\nalso available for most titles ( http://oreilly.com ). For more information, contact our corporate/institutional\\nsales department: 800-998-9938 or corporate@oreilly.com .\\nAcquisitions Editor:  Jessica Haberman\\nDevelopment Editor:  Gary O’Brien\\nProduction Editor:  Christopher FaucherInterior Designer:  David Futato\\nCover Designer:  Karen Montgomery\\nApril 2022:  First Edition\\nRevision History for the Early Release\\n2021-04-20: First Release\\n2021-05-07: Second Release\\nSee http://oreilly.com/catalog/errata.csp?isbn=9781098104597  for release details.\\nThe O’Reilly logo is a registered trademark of O’Reilly Media, Inc. Delta Lake: The Definitive  Guide , the\\ncover image, and related trade dress are trademarks of O’Reilly Media, Inc.\\nThe views expressed in this work are those of the authors, and do not represent the publisher’s views.\\nWhile the publisher and the authors have used good faith efforts to ensure that the information and\\ninstructions contained in this work are accurate, the publisher and the authors disclaim all responsibility\\nfor errors or omissions, including without limitation responsibility for damages resulting from the use of\\nor reliance on this work. Use of the information and instructions contained in this work is at your own\\nrisk. If any code samples or other technology this work contains or describes is subject to open source\\nlicenses or the intellectual property rights of others, it is your responsibility to ensure that your use\\nthereof complies with such licenses and/or rights.\\nThis work is part of a collaboration between O’Reilly and Databricks. See our statement of editorial inde‐\\npendence .Table of Contents\\n1.Basic Operations on Delta Lakes. . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . .  7\\nWhat is Delta Lake?                                                                                                           8\\nHow to start using Delta Lake                                                                                          9\\nUsing Delta Lake via local Spark shells                                                                       9\\nLeveraging GitHub or Maven                                                                                     10\\nUsing Databricks Community Edition                                                                     10\\nBasic operations                                                                                                               11\\nCreating your first Delta table                                                                                    11\\nUnpacking the Transaction Log                                                                                     14\\nWhat Is the Delta Lake Transaction Log?                                                                 16\\nHow Does the Transaction Log Work?                                                                     18\\nDealing With Multiple Concurrent Reads and Writes                                            30\\nOther Use Cases                                                                                                            35\\nDiving further into the transaction log                                                                     35\\nTable Utilities                                                                                                                    36\\nReview table history                                                                                                     36\\nVacuum History                                                                                                           37\\nRetrieve Delta table details                                                                                          39\\nGenerate a manifest file                                                                                               41\\nConvert a Parquet table to a Delta table                                                                    42\\nConvert a Delta table to a Parquet table                                                                    43\\nRestore a table version                                                                                                 43\\nSummary                                                                                                                           48\\n2.Time Travel with Delta Lake. . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . .  49\\nIntroduction                                                                                                                      49\\nUnder the hood of a Delta Table                                                                                    50\\nThe Delta Directory                                                                                                     50\\nvDelta Logs Directory                                                                                                    51\\nThe files of a Delta table                                                                                              54\\nTime Travel                                                                                                                       69\\nCommon Challenges with Changing Data                                                               69\\nWorking with Time Travel                                                                                          70\\nTime travel use cases                                                                                                    74\\nTime travel considerations                                                                                          82\\nSummary                                                                                                                           83\\n3.Continuous Applications with Delta Lake. . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . .  85\\nMake All Y our Streams Come True                                                                               86\\nSpark Streaming Was Built to Unify Batch and Streaming                                    87\\nExactly-Once Semantics                                                                                              92\\nPutting Some Structure Around Streaming                                                             94\\nStreaming with Delta                                                                                                   96\\nDelta as a Stream Source                                                                                             99\\nIgnore Updates and Deletes                                                                                        99\\nDelta Table as a Sink                                                                                                  102\\nAppendix                                                                                                                     103\\nvi | Table of ContentsCHAPTER 1\\nBasic Operations on Delta Lakes\\nA Note for Early Release Readers\\nWith Early Release ebooks, you get books in their earliest form—the authors’ raw and\\nunedited content as they write—so you can take advantage of these technologies long\\nbefore the official release of these titles.\\nThis will be the 2nd chapter of the final book. Please note that the GitHub repo will be\\nmade active later on.\\nIf you have comments about how we might improve the content and/or examples in\\nthis book, or if you notice missing material within this chapter, please reach out to the\\neditor at gobrien@oreilly.com .\\nIn the previous chapter, we discussed the progression of databases to data lakes, the\\nimportance of data reliability, and how to easily and effectively use Apache Spark™ to\\nbuild scalable and performant data processing pipelines. However, when building\\nthese data processing pipelines expressing the processing logic only solves half of the\\nend-to-end problem of building a pipeline. For a data practitioner, the ultimate goal\\nfor building pipelines is to efficiently query the processed data and get insights from\\nit. For a query to be efficient, the processed data must be saved in a storage format\\nand system such that the query engine can efficiently read the necessary (preferably\\nminimal) data to compute the query result. In addition, data is almost never static - it\\nhas to be continuously added, updated, corrected, and deleted when needed. It is\\nimportant that the storage system and the processing engine can work together to\\nprovide atomic transactionality  on the quality of the final result despite all the (possi‐\\nbly concurrent) read and write operations on the data.\\n7Concisely, Delta Lake is an open-source storage layer that brings ACID transactions\\nto Apache Spark™ and big data workloads. With Delta Lake providing atomic transac‐\\ntionality  for your data, this allows you - the data practitioner - to need to only focus\\non building your data processing pipelines by expressing the processing.\\nWhat is Delta Lake?\\nAs previously noted, over time, there have been different storage solutions built to\\nsolve this problem of data quality - from databases to data lakes. The transition from\\ndatabases to data lakes had the benefit of decoupling business logic from storage as\\nwell as the ability to independently scale compute and storage. But lost in this transi‐\\ntion was ensuring data reliability. Providing data reliability to data lakes led to the\\ndevelopment of Delta Lake.\\nBuilt by the original creators of Apache Spark, Delta Lake was designed to combine\\nthe best of both worlds for online analytical workloads (i.e., OLAP style): the transac‐\\ntional reliability of databases with the horizontal scalability of data lakes .\\nDelta Lake is a file-based, open-source storage format that provides ACID transac‐\\ntions, scalable metadata handling, and unifies streaming and batch data processing. It\\nruns on top of your existing data lakes and is compatible with Apache Spark and\\nother processing engines. Specifically, it provides the following features:\\nACID guarantees\\nDelta Lake ensures that all data changes written to storage are committed for\\ndurability and made visible to readers atomically. In other words, no more partial\\nor corrupted files! We will discuss more on the acid guarantees as part of the\\ntransaction log later in this chapter.\\nScalable data and metadata handling:\\nSince Delta Lake is built on data lakes, all reads and writes using Spark or other\\ndistributed processing engines are inherently scalable to petabyte-scale. However,\\nunlike most other storage formats and query engines, Delta Lake leverages Spark\\nto scale out all the metadata processing, thus efficiently handling metadata of bil‐\\nlions of files for petabyte-scale tables. We will discuss more on the transaction log\\nlater in this chapter.\\nAudit History and Time travel\\nThe Delta Lake transaction log records details about every change made to data\\nproviding a full audit trail of the changes. These data snapshots enable developers\\nto access and revert to earlier versions of data for audits, rollbacks, or to repro‐\\nduce experiments. We will dive further into this topic in Chapter 3: Time Travel\\nwith Delta.\\n8 | Chapter 1: Basic Operations on Delta LakesSchema enforcement and schema evolution\\nDelta Lake automatically prevents the insertion of data with an incorrect schema,\\ni.e. not matching the table schema. And when needed, it allows the table schema\\nto be explicitly and safely evolved to accommodate ever-change data. We will\\ndive further into this topic in Chapter 4 focusing on schema enforcement and\\nevolution.\\nSupport for deletes updates, and merge\\nMost distributed processing frameworks do not support atomic data modifica‐\\ntion operations on data lakes. Delta Lake supports merge, update, and delete\\noperations to enable complex use cases including but not limited to change-data-\\ncapture (CDC), slowly-changing-dimension (SCD) operations, and streaming\\nupserts. We will dive further into this topic in Chapter 5: Data modifications in\\nDelta.\\nStreaming and batch unification\\nA Delta Lake table has the ability to work both in batch and as a streaming source\\nand sink. The ability to work across a wide variety of latencies ranging from\\nstreaming data ingest to batch historic backfill to interactive queries all just work\\nout of the box. We will dive further into this topic in Chapter 6: Streaming Appli‐\\ncations with Delta.\\nFor more information, refer to the VLDB20 paper: Delta Lake: High-Performance\\nACID Table Storage over Cloud Object Stores .\\nHow to start using Delta Lake\\nDelta Lake is well integrated within the Apache Spark™ ecosystem so the easiest way\\nto start working with Delta Lake is to use it with Apache Spark. In this section, we\\nwill focus on three of the most straightforward mechanisms:\\n•Using Delta Lake via local Spark shells\\n•Leveraging GitHub or Maven\\n•Using Databricks Community Edition\\nUsing Delta Lake via local Spark shells\\nThe Delta Lake package is available as with the --packages  option when working\\nwith Apache Spark shells. This option is the simplest option when working with a\\nlocal version of Spark\\n# Using Spark Packages with PySpark shell\\n./bin/pyspark --packages io.delta:delta-core_2.12:1.0 \\n# Using Spark Packages with spark-shell\\nHow to start using Delta Lake | 9./bin/spark-shell --packages io.delta:delta-core_2.12:1.0 \\n# Using Spark Packages with spark-submit\\n./bin/spark-submit --packages io.delta:delta-core_2.12:1.0 \\n# Using Spark Packages with spark-shell\\n./bin/spark-sql --packages io.delta:delta-core_2.12:1.0 \\nLeveraging GitHub or Maven\\nIn this book, we will not focus on building standalone applications with Apache\\nSpark. For a good tutorial, please refer to Chapter 2 in Learning Spark 2nd Edition  or\\nbuilding your own self-contained application .\\nAs you build them, you can access Delta Lake via:\\n•GitHub: https://github.com/delta-io/delta\\n•Maven: https://mvnrepository.com/artifact/io.delta/delta-core\\nUsing Databricks Community Edition\\nAside from using your local machine or building your own self-contained applica‐\\ntion, you can try all of our examples in this book on the Databricks Community Edi‐\\ntion for free (Fig. 2-15). As a learning tool for Apache Spark, the Community Edition\\nhas many tutorials and examples worthy of note. As well as writing your own note‐\\nbooks in Python, R, Scala, or SQL, you can also import other notebooks, including\\nJupyter notebooks.\\nFigure 1-1. Databricks Community Edition\\n10 | Chapter 1: Basic Operations on Delta LakesTo get an account, go to https://databricks.com/try , and follow the instructions to use\\nthe Community Edition for free.\\nBasic operations\\nIn this section, we will cover the basic operations of building Delta tables. As of writ‐\\ning this, Delta has full APIs in three languages commonly used in the big data ecosys‐\\ntem: SQL, Scala, and Python. It can store data in file systems like HDFS and cloud\\nobject stores including S3 and ADLS gen2. It is designed to be written primarily by\\nSpark applications and can be read by many open-source data engines like Spark\\nSQL, Hive, Presto (or Trino), Ballista (Rust), and several enterprise products like\\nAWS Athena, Azure Synapse, BigQuery, and Dremio.\\nCreating your first  Delta table\\nLet’s create our first Delta table! Like databases, to create our Delta table we can first\\ncreate a table definition and define the schema or, like data lakes simply write a Spark\\nDataFrame to storage in the Delta format.\\nWriting your Delta table\\nWhen creating a Delta table, you are writing files to some storage (e.g. file system,\\ncloud object stores). All of the files together stored in a directory of a particular struc‐\\nture (more on this later) make up your table. Therefore, when we create a Delta table,\\nwe are in fact writing files to some storage location. For example, the following Spark\\ncode snippet takes an existing Spark DataFrame and writes it in the Apache Parquet\\nstorage format in the folder location /data  using the Spark DataFrame API.\\n              dataframe.write.format(\"parquet\").save(\"/data\")\\n            \\nFor more information on the Spark DataFrame API, a good refer‐\\nence is Learning Spark 2nd Edition .\\nWith one simple modification of the preceding code snippet, you can now create a\\nDelta table.\\nBasic operations | 11Figure 1-2. Instead of parquet, simply say delta\\nLet’s see how that looks in code by using Apache Spark to create a Delta table from a\\nDataFrame. In the following code example, we will first create the Spark DataFrame\\ndata and then use the write method to save the table to storage.\\n%python\\n# Create data DataFrame\\ndata = spark.range(0, 5)\\n# Write the data DataFrame to /delta location\\ndata.write.format(\"delta\").save(\"/delta\" )\\n%scala\\n// Create data DataFrame\\nval data = spark.range(0, 5)\\n// Write the data DataFrame to /delta location\\ndata.write.format(\"delta\").save(\"/delta\" )\\nIt is important to note that in most production environments when you are working\\nwith large amounts of data, it is important to partition your data. The following\\nexample partitions your data by date (a commonly followed best practice):\\n%python\\n# Write the Spark DataFrame to Delta table partitioned by date\\ndata.write.partitionBy (\"date\").format(\"delta\").save(\"/delta\" )\\n%scala\\n// Write the Spark DataFrame to Delta table partitioned by date\\ndata.write.partitionBy (\"date\").format(\"delta\").save(\"/delta\" )\\nIf you already have an existing Delta table and would like to append or overwrite data\\nto your table, include the mode  method in your statement.\\n%python\\n># Append new data to your Delta table\\ndata.write.format(\"delta\").mode(\"append\" ).save(\"/delta\" )\\n# Overwrite your Delta table\\ndata.write.format(\"delta\").mode(\"overwrite\" ).save(\"/delta\" )\\n%scala\\n// Append new data to your Delta table\\n12 | Chapter 1: Basic Operations on Delta Lakesdata.write.format(\"delta\").mode(\"append\" ).save(\"/delta\" )\\n// Overwrite your Delta table\\ndata.write.format(\"delta\").mode(\"overwrite\" ).save(\"/delta\" )\\nReading your Delta table\\nSimilar to writing your Delta table, you can use the DataFrame API to read the same\\nfiles from your Delta table.\\n%python\\n# Read the data DataFrame from the /delta location\\nspark.read.format(\"delta\").load(\"/delta\" ).show()\\n%scala\\n// Read the data DataFrame from the /delta location\\nspark.read.format(\"delta\").load(\"/delta\" ).show()\\nY ou can also read the table using SQL by specifying the file location after specifying\\ndelta .\\n%sql\\nSELECT * FROM delta.`/delta`\\nThe output of this table can be seen below.\\n+---+\\n| id|\\n+---+\\n|  4|\\n|  2|\\n|  3|\\n|  0|\\n|  1|\\n+---+\\nReading your metastore defined  Delta table\\nIn the previous sections, we have been reading our Delta tables directly from the file\\nsystem. But how would you read a metastore defined Delta table? For example,\\ninstead of reading our Delta table in SQL using the file path:\\n%sql\\nSELECT * FROM delta.`/delta`\\nhow would you read using a metastore-defined table such as:\\n%sql\\nSELECT * FROM myTable\\nTo do this, you would need to first define the table within the metastore using the\\nsaveAsTable  method or a CREATE TABLE  statement.\\nBasic operations | 13%python\\n# Write the data DataFrame to the metastore defined as myTable\\ndata.write.format(\"delta\").saveAsTable (\"myTable\" )\\n%scala\\n// Write the data DataFrame to the metastore defined as myTable\\ndata.write.format(\"delta\").saveAsTable (\"myTable\" )\\nNote, when using the saveAsTable  method, you will save the Delta table files into a\\nlocation managed by the metastore (e.g. /user/hive/warehouse/myTable ). If you\\nwant to use SQL or control the location of your Delta table, then you initially save it\\nusing the save  method where you specify the location (e.g. /delta ) and then create\\nthe table using the following SQL statements.\\n%sql\\n-- Create a table in the metastore\\nCREATE TABLE myTable (\\n  id INTEGER)\\nUSING DELTA\\nLOCATION  \"/delta\"\\nAs noted previously, it will be important to partition your large tables.\\n%sql\\nCREATE TABLE id (\\n  date DATE\\n  id INTEGER)\\nUSING DELTA\\nPARTITION  BY date\\nLOCATION  \"/delta\"\\nNote, the LOCATION  property that points to the underlying files that make up your\\nDelta table.\\nSide Note: What is a metastore?\\nGenerally in the Hadoop ecosystem, to reference a dataset via the flavors of SQL\\n(Spark SQL, Hive, Presto, Impala, etc), you need to make a table definition in a meta‐\\nstore  (commonly this is a Hive metastore). This table definition is a metadata entry\\nthat will describe to data processing frameworks such as Apache Spark the data loca‐\\ntion, storage format, table schema, as well as other properties.\\nUnpacking the Transaction Log\\nIn the previous section where we discuss the basic operations of Delta Lake, many\\nreaders may have been under the impression that there was more to ensuring data\\nreliability from a usability  or user experience  perspective. The basic operations of\\nDelta Lake appear to be simply Spark SQL statements. We have alluded to improved\\n14 | Chapter 1: Basic Operations on Delta Lakesfunctionality such as time travel (chapter 3) and DML operations (chapter 5) but this\\nmay not seem apparent from those basic operations. But the real benefit of Delta Lake\\nis that the operations themselves provide ACID transactional protection. And to\\nunderstand how Delta Lake provides those protections, we need to first unpack and\\nunderstand the transaction log.\\nFor example, when running this notebook in Databricks Community Edition, note\\nthat there is little difference between the Parquet and Delta tables generated from the\\nsame source.\\nFigure 1-3. Difference  between parquet and Delta tables\\nThe key difference is the _delta_log  folder which is the Delta transaction log. This\\ntransaction log is key to understanding Delta Lake because it is the underlying infra‐\\nstructure for many of its most important features including but not limited to ACID\\ntransactions, scalable metadata handling, and time travel. In this section, we’ll explore\\nUnpacking the Transaction Log | 15what the Delta Lake transaction log is, how it works at the file level, and how it offers\\nan elegant solution to the problem of multiple concurrent reads and writes.\\nWhat Is the Delta Lake Transaction Log?\\nThe Delta Lake transaction log (also known as the Delta Log ) is an ordered record of\\nevery change that has ever been performed on a Delta Lake table since its inception.\\nSingle Source of Truth\\nDelta Lake is built on top of Apache Spark™ in order to allow multiple readers and\\nwriters of a given table to all work on the table at the same time. To show users cor‐\\nrect views of the data at all times, the Delta Lake transaction log serves as a single\\nsource of truth – the central repository that tracks all changes that users make to the\\ntable. Y ou could compare this transaction log feature in Delta Lake as a single source\\nof truth to the .git directory in a git-managed source code repository.\\nThe concept of a single source of truth is important because, over time, processing\\njobs will fail in your data lake. The result is that processing jobs or tasks will leave\\npartial files that do not get cleaned up. Subsequent processing or queries will not be\\nable to ascertain which files should or should not be included in their queries.\\nFigure 1-4. Partial file example\\nFor example, the preceding diagram represents a common data lake processing sce‐\\nnario.\\nTime\\nPeriodTable\\n16 | Chapter 1: Basic Operations on Delta Lakest0 The table is represented by 2 Parquet files\\nt1 A processing job ( job 1 ) extracts files  3 and 4 and writes them to disk. But due to some error (e.g. network\\nhiccup, storage temporary offline,  etc.), an incomplete part of file 3 and none of file 4 are written to storage.\\nThere is no automated mechanism to clean up this data; this partial of 3.parquet  can be queried within your\\ndata lake.\\nt2 A new version of the same processing job ( job 1 v2 ) is executed and this time it completes its job successfully\\nwith 3\\'.parquet  and 4.parquet . But because the partial 3.parquet  exists alongside with 3\\'.par\\nquet  there will be double counting.\\nBy using a transaction log to track which files are valid, we can avoid the preceding\\nscenario. Thus, when a user reads a Delta Lake table for the first time or runs a new\\nquery on an open table that has been modified since the last time it was read, Spark\\nchecks the transaction log to see what new transactions have posted to the table, and\\nthen updates the end user’s table with those new changes. This ensures that a user’s\\nversion of a table is always synchronized with the master record as of the most recent\\nquery and that users cannot make divergent, conflicting changes to a table.\\nLet’s repeat the same partial file example except for this time we run this on a Delta\\ntable with the Delta transaction log.\\nFigure 1-5. Partial file example with Delta transaction log\\nTime\\nPeriodTable Transaction Log\\nUnpacking the Transaction Log | 17t0 The table is represented by 2 Parquet files Version 0: The transaction log records the two files  that\\nmake up the Delta table.\\nt1 A processing job ( job 1 ) extracts files  3 and 4 and\\nwrites them to disk. But due to some error (e.g.\\nnetwork hiccup, storage temporary offline,  etc.), an\\nincomplete part of file 3 and none of file 4 are written\\nto storage. There is no automated mechanism to clean\\nup this data; this partial of 3.parquet  can be\\nqueried within your data lake.Because the job fails, the transaction was NOT committed\\nto the transaction log. No new files  are recorded.\\nAny queries against the Delta table at this time will be\\nprovided a list of the initial two files  and Spark will only\\nquery these two files  even if other files  are in the folder.\\nt2 A new version of the same processing job ( job 1 \\nv2) is executed and this time it completes its job\\nsuccessfully with 3\\'.parquet  and 4.parquet .\\nBut because the partial 3.parquet  exists alongside\\nwith 3\\'.parquet  there will be double counting.Version 1: Because the job completes successfully, a new\\ntransaction is committed with the two new files.  Thus\\nany queries against the table will be provided with these\\nfour files.  Because a new transaction is committed, it is\\nsaved as a new version (i.e. V1 instead of V0).\\nThe exact same actions happened as in the previous example but this time the Delta\\ntransaction log provides atomicity  and ensures data reliability .\\nThe Implementation of Atomicity on Delta Lake\\nOne of the four properties of ACID transactions, atomicity, guarantees that opera‐\\ntions (like an INSERT or UPDATE) performed on your data lake either complete\\nfully or don’t complete at all. Without this property, it’s far too easy for a hardware\\nfailure or a software bug to cause data to be only partially written to a table, resulting\\nin messy or corrupted data.\\nThe transaction log is the mechanism through which Delta Lake is able to offer the\\nguarantee of atomicity - if it’s not recorded in the transaction log, it never happened.\\nBy only recording transactions that execute fully and completely, and using that\\nrecord as the single source of truth, the Delta Lake transaction log allows users to rea‐\\nson about their data, and have peace of mind about its fundamental trustworthiness,\\nat petabyte scale.\\nHow Does the Transaction Log Work?\\nThis section provides internals of how the Delta transaction log works.\\nBreaking Down Transactions Into Atomic Commits\\nWhenever a user performs an operation to modify a table (e.g., INSERT , DELETE ,\\nUPDATE  or MERGE ), Delta Lake breaks that operation down into a series of discrete\\nsteps composed of one or more of the actions below.\\nUpdate metadata\\nUpdates the table’s metadata including but not limited to changing the table’s\\nname, schema or partitioning.\\n18 | Chapter 1: Basic Operations on Delta LakesAdd file\\nadds a data file to the transaction log.\\nRemove file\\nremoves a data file from the transaction log.\\nSet transaction\\nRecords that a structured streaming job has committed a micro-batch with the\\ngiven ID.\\nChange protocol\\nenables new features by switching the Delta Lake transaction log to the newest\\nsoftware protocol.\\nCommit info\\nContains information around the commit, which operation was made, from\\nwhere, and at what time.\\nThose actions are then recorded in the transaction log as ordered, atomic units\\nknown as commits.\\nDelta Transaction Log Protocol\\nIn this section, we describe how the Delta transaction log brings ACID  properties to\\nlarge collections of data, stored as files, in a distributed file system or object-store.\\nThe protocol was designed with the following goals in mind:\\nSerializable ACID Writes\\nmultiple writers can concurrently modify a Delta table while maintaining ACID\\nsemantics. (Watch this video  on how each concurrent writes are handled)\\nSnapshot Isolation for Reads\\nreaders can read a consistent snapshot of a Delta table, even when multiple writ‐\\ners are concurrently writing.\\nScalability to billions of partitions or files\\nqueries against a Delta table can be planned on a single machine or in parallel.\\nSelf-describing\\nall metadata for a Delta table is stored alongside the data. This design eliminates\\nthe need to maintain a separate metastore just to read the data and also allows\\nstatic tables to be copied or moved using standard filesystem tools.\\nSupport for incremental processing\\nreaders can tail the Delta log to determine what data has been added in a given\\nperiod of time, allowing for efficient streaming.\\nUnpacking the Transaction Log | 19Logstore\\nThink about the existence of the delta files for a second. The logs, versions, and files\\nthat are being generated must exist somewhere, some system or store for files. Log‐\\nStore is the general interface for all critical file system operations required to read and\\nwrite the Delta transaction log. Because most storage systems do not provide atomic‐\\nity guarantees out-of-the-box, Delta Lake transactional operations go through the\\nLogStore API instead of accessing the storage system directly.\\n•Any file written through this store must be made visible, atomically. In other\\nwords, it should be visible in its entirety or not visible at all. It should not gener‐\\nate partial files.\\n•Only one writer must be able to create a file at the final destination. This is\\nbecause many processes can occur simultaneously and to ensure decent speed,\\nmany writers write to their own files in parallel.\\n•The Logstore offers ACID consistent listing of files\\nThe Delta Lake Transaction Log at the File Level\\nAs noted previously, when a Delta table is created, that table’s transaction log is auto‐\\nmatically created in the _delta_log  subdirectory. As they make changes to that table,\\nthose changes are recorded as ordered, atomic commits in the transaction log. Each\\ncommit is written out as a JSON file, starting with 000000.json . Additional changes\\nto the table generate subsequent JSON files in ascending numerical order so that the\\nnext commit is written out as 000001.json , the following as 000002.json , and so on.\\nEach numeric JSON file increment represents a new version of the table.\\nFigure 1-6. Delta on disk\\nImplementing Atomicity.    An important thing to understand is the transaction log is\\nthe single source of truth for your Delta table. So any reader that’s reading through\\n20 | Chapter 1: Basic Operations on Delta Lakesyour Delta table, will take a look at the transaction log first. Therefore changes to the\\ntable are stored as ordered atomic units called commits.\\nTo continue the above example, we add additional records to our Delta table from the\\ndata files 1.parquet  and 2.parquet . That transaction would automatically be added\\nto the transaction log, saved to disk as commit  000000.json . Then, perhaps we\\nchange our minds and decide to remove those files (e.g., run a DELETE  from table\\noperation) and add a new file instead ( 3.parquet ) through an INSERT  operation.\\nThose actions would be recorded as the next commit - and table version - in the\\ntransaction log, as 000001.json , as shown in the following diagram.\\nFigure 1-7. Implementing atomicity in Delta\\nEven though 1.parquet  and 2.parquet  are no longer part of our Delta Lake table,\\ntheir addition and removal are still recorded in the transaction log because those\\noperations were performed on our table – despite the fact that they ultimately can‐\\nceled each other out. Delta Lake still retains atomic commits like these to ensure that\\nin the event we need to audit our table or use “time travel” (to discuss in chapter 3) to\\nsee what our table looked like at a given point in time, we could do so accurately.\\nTo provide context, let’s continue with our notebook and read the JSON file that\\nmakes up the first transaction (version 1) inserting data into our Delta table.\\n>%python\\n# Read the transaction log of version 1\\nj0 = spark.read.json(\"/.../000001.json\" )\\nUnpacking the Transaction Log | 21%scala\\n// Read the transaction log of version 1\\nval j0 = spark.read.json(\"/.../000001.json\" )\\nFigure 1-8. Reviewing the transaction log structure\\nThe transaction contains many pieces of information stored as strings or columns\\nwithin the transaction log JSON. Let’s focus on the commit , add, and CRC pieces.\\nCommit Information.    The Delta transaction log commits metadata can be read using\\nthe following code snippets.\\n%python\\n# Commit Information\\ndisplay(j0.select(\"commitInfo\" ).where(\"commitInfo is not null\" ))\\n22 | Chapter 1: Basic Operations on Delta Lakes%scala\\n// Commit Information\\ndisplay(j0.select(\"commitInfo\" ).where(\"commitInfo is not null\" ))\\nThe following is a text version of the JSON output read from the preceding com‐\\nmand.\\nclusterId: \"0127-045215-pined152\"\\nisBlindAppend: true\\nisolationLevel: \"WriteSerializable\"\\nnotebook: {\"notebookId\": \"8476282\"}\\noperation: \"STREAMING UPDATE\"\\noperationMetrics: {\"numAddedFiles\": \"1\", \"numOutputBytes\": \"492\", \"numOutpu-\\ntRows\": \"0\", \"numRemovedFiles\": \"0\"}\\noperationParameters: {\"epochId\": \"0\", \"outputMode\": \"Append\", \"queryId\": \\n\"892c1abd-581f-4c0f-bbe7-2c386feb3dbd\"}\\nreadVersion: 0\\ntimestamp: 1603581133495\\nuserId: \"100599\"userName: \" denny[dot]lee[@]databricks.com \"\\nThe metadata contains a lot of interesting information, but let’s focus on the ones\\naround the file system.\\nmetadata description\\noperation The type of operation that is happening, in this case, data was inserted via Spark Structured streaming\\nwriteStream job (i.e. STREAMING UPDATE )\\noperationMetrics Notes how many files  were added ( numAddedFiles ), removed ( numRemovedFiles ), output\\nrows (numOutputRows ), and output bytes ( numOutputBytes )\\noperationParameters From the files  perspective, whether this operation would append  or overwrite  the data within the\\ntable.\\nreadVersion The table version associated with this transaction commit.\\nclusterID, notebook Identifies  which Databricks cluster and notebook that executed this commit\\nuserID, userName ID and name of the user executing this operation\\nAdd Information.    The Delta transaction log add metadata can be read using the fol‐\\nlowing code snippets.\\n%python\\n# Commit Information\\ndisplay(j0.select(\"add\").where(\"add is not null\" ))\\n%scala\\n// Commit Information\\ndisplay(j0.select(\"add\").where(\"add is not null\" ))\\nThe following is a text version of the JSON output read from the preceding com‐\\nmand.\\ndataChange:  true\\nmodificationTime:  1603581134000\\nUnpacking the Transaction Log | 23path: \"part-00000-95648a41-bf33-4b67-9979-1736d07bef0e-c000.snappy.parquet\"\\nsize: 492\\nstats: \"{\\\\\"numRecords\\\\\":0,\\\\\"minValues\\\\\":{},\\\\\"maxValues\\\\\":{},\\\\\"nullCount\\\\\":{}}\"\\nThe metadata is particularly important in that it lists the file that makes up the table\\nversion.\\nmetadata description\\npath A list of the file(s)  added to a Delta table per a committed transaction\\nsize The size of the file(s)\\nstats The statistics stored in the transaction log to help the reader understand the size and scope of data files  that need\\nto be read.\\nTypically when reading files from storage, Spark (and other distributed processing\\nframeworks) will simply read all of the files in a folder. Prior to reading the files, it\\nmust first list the files ( listFrom ) which can be extremely inefficient especially on\\ncloud object stores.\\nBut in the case of Delta Lake, the files are listed directly within the transaction log\\nitself. So instead of Spark (or any other processing framework that can read the Delta\\ntransaction log), listing all the files, the transaction log provides all of the files that are\\nassociated with a table version. Therefore Spark can just query the files directly\\nthrough their individual paths which can have a significant performance improve‐\\nment than listFrom  especially for internet-scale systems with petabytes of data.\\nCRC file.    For each transaction, there is both a JSON file as well as a CRC file. This file\\ncontains key statistics for the table version (i.e. transaction) allowing Delta Lake to\\nhelp Spark optimize its queries.\\nTo review this data, let’s just review the file directly from the file system using the fol‐\\nlowing command.\\n                %sh head /dbfs/ml/loan_by_state/_delta_log/\\n00000000000000000001.crc\\n              \\nBelow are the results from this query\\n{\\n\"tableSizeBytes\":1855,\\n\"numFiles\":2,\\n\"numMetadata\":1,\\n\"numProtocol\":1,\\n\"numTransactions\":1\\n}\\nThe two most notable pieces of metadata are:\\n24 | Chapter 1: Basic Operations on Delta LakestableSizeInBytes\\nThe table size in bytes so Spark and Delta Lake can optimize their queries\\nnumFiles\\nImportant for Spark especially in scenarios like dynamic partition pruning\\nAs you can read from the preceding high-level primer, the Delta Lake transaction log\\ntracks the files and other metadata to ensure both atomic transactions  and data relia‐\\nbility .\\nSpark does not eagerly remove the files from disk, even though we\\nremoved the underlying data files from our table. Users can delete\\nthe files that are no longer needed using V ACUUM  (more on this\\nlater under the Table Utilities section).\\nQuickly Recomputing State With Checkpoint Files\\nPrior to this, our focus is that of a single transaction log in the form of JSON files. But\\nfor large-scale systems or any streaming system, this would result in creating the\\n“small-file” problem where it becomes ever more inefficient to query the transaction\\nlog folder (i.e. _delta_log  subdirectory).\\nTo alleviate this issue, Delta Lake creates a checkpoint file in Parquet format after it\\ncreates the 10th commits (i.e. transaction).\\nFigure 1-9. Create transaction log after  every 10th commit\\nThese checkpoint files save the entire state of the table at a point in time – in native\\nParquet format that is quick and easy for Spark to read. In other words, they offer the\\nUnpacking the Transaction Log | 25Spark reader a sort of “shortcut” to fully reproducing a table’s state that allows Spark\\nto avoid reprocessing what could be thousands of tiny, inefficient JSON files.\\nWhen recomputing the state of the table, Spark will read and cache the available\\nJSON files that make up the transaction log. For example, if there have been only\\nthree committed operations or commits to the table (including the table creation),\\nSpark will read all three files and cache the results into memory (i.e. cache version 2).\\nFigure 1-10. Spark caching after  the third commit\\nInstead of continually reading the transaction log for this information, all of the\\nSpark readers requesting data from this table can simply reference the cached copy of\\nDelta’s state. As more commits are performed against the Delta table, more JSON files\\nwill be added to the _delta_log  folder.\\n26 | Chapter 1: Basic Operations on Delta LakesFigure 1-11. Spark caching after  the eight commit\\nTo continue this example, let’s say this table five additional commits, then Spark will\\ncache version 7 of the data. Note, at this point, Spark will list all of the transactions\\nfrom version 0 instead of reading from version 2 to ensure that earlier transactions\\nhave completed.\\nUnpacking the Transaction Log | 27Figure 1-12. Spark caching after  the 13th commit including checkpoint creation\\nAs noted earlier, Delta Lake will create a checkpoint file ( 0000010.checkpoint.par\\nquet ) after the 10th commit. Delta Lake will still listFrom version 0 to avoid late\\ntransactions and cache version 12.\\nTo read the checkpoint file, run the following statement.\\n%python\\n# Review the transaction log checkpoint file\\nchkpt0 = spark.read.parquet(\"/ml/loan_by_state/_delta_log/000010.check\\npoint.parquet\" )\\n%scala\\n// Review the transaction log checkpoint file\\nval chkpt0 = spark.read.parquet(\"/ml/loan_by_state/_delta_log/000010.check\\npoint.parquet\" )\\nThe metadata you will read is a union of all of the previous transactions. This\\nbecomes apparent when you read the query for the add information.\\n%python\\n# Add Information\\ndisplay(chkpt0.select(\"add\").where(\"add is not null\" ))\\n%scala\\n// Add Information\\ndisplay(chkpt0.select(\"add\").where(\"add is not null\" ))\\n28 | Chapter 1: Basic Operations on Delta LakesFigure 1-13. Add information in the checkpoint file\\nIn addition to the checkpoint file being in Parquet format (thus Spark can read it\\neven faster) and containing all of the transactions prior to it, notice how the stats  in\\nthe original JSON file were in string format.\\nstats: \"{\\\\\"numRecords\\\\\":7,\\\\\"minValues\\\\\":{\\\\\"addr_state\\\\\":\\\\\"IA\\\\\",\\\\\"count\\\\\":\\n3,\\\\\"stream_no\\\\\":3},\\\\\"maxValues\\\\\":{\\\\\"addr_state\\\\\":\\\\\"TX\\\\\",\\\\\"count\\\\\":9,\\\\\"stream_no\\n\\\\\":3},\\\\\"nullCount\\\\\":{\\\\\"addr_state\\\\\":0,\\\\\"count\\\\\":0,\\\\\"stream_no\\\\\":0}}\"\\n            \\nAs part of the checkpoint creation process, there is a stats_parsed  column that con‐\\ntains the statistics as nested columns instead of strings.\\nstats_parsed:{\\n\"numRecords\": 7, \\n\"minValues\": {\\n\"addr_state\": \"IA\", \\n\"count\": 3, \\n\"stream_no\": 3\\n}, \\n\"maxValues\": {\\n\"addr_state\": \"TX\", \\n\"count\": 9, \\n\"stream_no\": 3\\nUnpacking the Transaction Log | 29}, \\n\"nullCount\": {\\n\"addr_state\": 0, \\n\"count\": 0, \\n\"stream_no\": 0\\n}\\n}\\nBy using nested columns instead of strings, Spark can read the statistics significantly\\nfaster especially when it needs to read all of the files created for petabyte-scale data\\nlakes.\\nFigure 1-14. Spark caching after  the 15th commit including checkpoint creation\\nAs it is important to note that once the Parquet checkpoint file has been created, sub‐\\nsequent listFrom calls are from the checkpoint file instead of going back to version 0.\\nThus after the 15 commits, Delta will cache version 14 but need only to listFrom the\\nversion 10 checkpoint file.\\nDealing With Multiple Concurrent Reads and Writes\\nNow that we understand how the Delta Lake transaction log works at a high level, let’s\\ntalk about concurrency. So far, our examples have mostly covered scenarios in which\\nusers commit transactions linearly, or at least without conflict. But how does Delta\\nLake deal with multiple concurrent reads and writes? Because Delta Lake is powered\\nby Apache Spark, the expectation is that multiple users concurrently modify a single\\ntable. To do this, Delta Lake employs optimistic concurrency control.\\nWhat Is Optimistic Concurrency Control?\\nOptimistic concurrency control is a method of dealing with concurrent transactions\\nthat assumes that transactions made to a table by different users can complete\\nwithout conflicting with one another. It is incredibly fast because when dealing with\\npetabytes of data, there’s a high likelihood that users will be working on different\\n30 | Chapter 1: Basic Operations on Delta Lakesparts of the data altogether, allowing them to complete non-conflicting transactions\\nsimultaneously.\\nWhen working with tables, as long as different clients are modifying different parts of\\nthe table (e.g. different partitions) or performing actions that do not conflict (e.g. two\\nclients reading from the table), those operations do not conflict so we can optimisti‐\\ncally let them all complete their task. But for situations where clients modify the same\\nparts of the table concurrently, Delta Lake has a protocol to resolve this.\\nSolving conflicts  optimistically\\nEnsuring serializability.    Another key piece that Delta requires for consistent guaran‐\\ntees is mutual exclusion . We need to agree on the order of changes, even when there\\nare multiple writers and this provides a guarantee in databases called serializability .\\nThe fact that even though things are happening concurrently, you could play them as\\nif they happened in a synchronous ordered manner.\\nFor example, we have user 1 reading 000000.json\\nFigure 1-15. User 1 reading\\nAnd we have user2 that reads 000001.json\\n.\\nFigure 1-16. Users 1 and 2 reading the transaction log concurrently\\nUnpacking the Transaction Log | 31As you’re trying to commit 000002.json , user2 two wins and user1 takes a look and\\nsees that, 000002.json  is already there. Due to this requirement of mutual exclusion,\\nit will have to say, “Oh this commit failed” and let me try to commit 000003.json\\ninstead.\\nFigure 1-17. Ensuring serializability\\nApplying Optimistic Concurrency Control in Delta.    In order to offer ACID transactions,\\nDelta Lake has a protocol for figuring out how commits should be ordered (known as\\nthe concept of serializability  in databases) and determining what to do in the event\\nthat two or more commits are made at the same time.\\nDelta Lake handles these cases by implementing a rule of mutual exclusion , then\\nattempting to solve any conflict optimistically. This protocol allows Delta Lake to\\ndeliver on the ACID principle of isolation , which ensures that the resulting state of\\nthe table after multiple, concurrent writes is the same as if those writes had occurred\\nserially, in isolation from one another.\\nIn general, the process proceeds like this:\\n1.Record the starting table version.\\n2.Record reads/writes.\\n3.Attempt a commit.\\n4.If someone else wins, check whether anything you read has changed.\\n5.Repeat.\\nFor example, if two users read from the same table and then both attempt to insert\\ndata at the same time.\\n32 | Chapter 1: Basic Operations on Delta LakesFigure 1-18. Solving conflicts  optimistically\\n•Delta Lake records the starting table version of the table (version 0) that is read\\nprior to making any changes.\\n•Users 1 and 2 both attempt to append some data to the table at the same time.\\nHere, we’ve run into a conflict because only one commit can come next and be\\nrecorded as 000001.json .\\n•Delta Lake handles this conflict with the concept of mutual exclusion, i.e. only\\none user can successfully commit 000001.json . User 1’s commit is accepted,\\nwhile User 2’s is rejected.\\n•Rather than throw an error for User 2, Delta Lake prefers to handle this conflict\\noptimistically. It checks to see whether any new commits have been made to the\\ntable, and updates the table silently to reflect those changes, then simply retries\\nUser 2’s commit on the newly updated table (without any data processing), suc‐\\ncessfully committing 000002.json .\\nIn the vast majority of cases, this reconciliation happens silently, seamlessly, and suc‐\\ncessfully. However, in the event that there’s an irreconcilable problem that Delta Lake\\ncannot solve optimistically (for example, if User 1 deleted a file that User 2 also\\ndeleted), the only option is to throw an error.\\nAs a final note, since all of the transactions made on Delta Lake tables are stored\\ndirectly to storage, this process satisfies the ACID property of durability , meaning it\\nwill persist even in the event of system failure.\\nMultiversion Concurrency Control.    How all of this works within the file system is that\\nDelta’s transactions are implemented using Multiversion Concurrency Control\\n(MVCC). It is a concurrency control method commonly used by relational database\\nmanagement systems to provide concurrent access to the database (and its tables)\\nUnpacking the Transaction Log | 33within the context of transactions. As Delta Lake’s data objects and log are immuta‐\\nble, Delta Lake utilizes MVCC to both protect existing data, i.e. provides transac‐\\ntional guarantees between writes, as well as speed up query and write performance. It\\nalso has the benefit of making it straightforward to query a past snapshot of the data,\\nas is common in MVCC implementations.\\nUnder this mechanism, writes operate in three stages:\\nRead\\nFirst the system reads the latest available version of the table to identify which\\nrows need to be modified.\\nWrite\\nStages all the changes by writing new data files. Note, all changes whether inserts\\nor modifications are in the form of writing new files.\\nValidate and commit\\nBefore committing the changes, it checks whether the proposed changes conflict\\nwith any other changes that may have been concurrently committed since the\\nsnapshot that was read. If there are no conflicts, all the staged changes are com‐\\nmitted as a new versioned snapshot, and the write operation succeeds. However,\\nif there are conflicts, the write operation fails with a concurrent modification\\nexception rather than corrupting the table as would happen with the write opera‐\\ntion on a Parquet table.\\nAs a table changes, Delta’s MVCC algorithm keeps multiple copies of the data around\\nrather than immediately replacing files that contain records that are being updated or\\nremoved. MVCC allows serializability and snapshot isolation for consistent views of a\\nstate of a table so that the readers continue to see a consistent snapshot view of the\\ntable that the Apache Spark job started with, even when a table is modified during a\\njob. They can efficiently query a snapshot by using the transaction log to selectively\\nchoose which data files to process when there are concurrent modifications being\\ndone by writers.\\nTherefore, writers modify a table in two phases:\\n•First, they optimistically write out new data files or updated copies of existing\\nones.\\n•Then, they commit, creating the latest atomic version of the table by adding a\\nnew entry to the log. In this log entry, they record which data files to logically add\\nand remove, along with changes to other metadata about the table.\\n34 | Chapter 1: Basic Operations on Delta LakesOther Use Cases\\nThere are many interesting use cases that can be implemented because of Delta Lake’s\\ntransaction log. Below are a couple of examples that we will dive into in chapter 3.\\nTime Travel\\nEvery table is the result of the sum total of all of the commits recorded in the Delta\\nLake transaction log – no more and no less. The transaction log provides a step-by-\\nstep instruction guide, detailing exactly how to get from the table’s original state to its\\ncurrent state.\\nTherefore, we can recreate the state of a table at any point in time by starting with an\\noriginal table, and processing only commits made prior to that point. This powerful\\nability is known as “time travel, ” or data versioning, and can be a lifesaver in any\\nnumber of situations. For more information, read the blog post Introducing Delta\\nTime Travel for Large Scale Data Lakes , or refer to the Delta Lake time travel docu‐\\nmentation .\\nData Lineage and Debugging\\nAs the definitive record of every change ever made to a table, the Delta Lake transac‐\\ntion log offers users a verifiable data lineage that is useful for governance, audit, and\\ncompliance purposes. It can also be used to trace the origin of an inadvertent change\\nor a bug in a pipeline back to the exact action that caused it. Users can run\\nDESCRIBE HISTORY  to see metadata around the changes that were made.\\nDiving further into the transaction log\\nIn this section, we dove into the details of how the Delta Lake transaction log works,\\nincluding:\\n•What the transaction log is, how it’s structured, and how commits are stored as\\nfiles on disk.\\n•How the transaction log serves as a single source of truth, allowing Delta Lake to\\nimplement the principle of atomicity.\\n•How Delta Lake computes the state of each table – including how it uses the\\ntransaction log to catch up from the most recent checkpoint.\\n•Using optimistic concurrency control to allow multiple concurrent reads and\\nwrites even as tables change.\\n•How Delta Lake uses mutual exclusion to ensure that commits are serialized\\nproperly, and how they are retried silently in the event of a conflict.\\nFor more information, refer to:\\nUnpacking the Transaction Log | 35•Diving into Delta Lake: Unpacking the Transaction Log  (blog)\\n•Diving into Delta Lake: Unpacking the Transaction Log  (video tech talk)\\n•Diving into Delta Lake: Unpacking the Transaction Log v2  (Data + AI Summit\\nEU 2020 session)\\nTable Utilities\\nIf there is a central theme for reviewing the transaction log, MVCC, optimistic con‐\\ncurrency control, and the underlying file system when working with Delta Lake is\\nthat it is about the manipulation of the underlying files that make up the Delta table.\\nIn this section, we provide an overview of the various table utilities to simplify opera‐\\ntions.\\nReview table history\\nY ou can retrieve information on the operations, user, timestamp, and so on for each\\nwrite to a Delta table by running the history command. This history will tell us things\\nlike what type of write occurred (append, merge, delete, etc.), was it a blind append,\\nwas it restricted to a specific partition, and provides operational metrics on the\\namount of data written. All of this information is interesting, but the operational met‐\\nrics provide the most insight into how your table is changing, showing how many\\nfiles, rows, and bytes were added or removed.\\nNote, the operations are returned in reverse chronological order and by default, table\\nhistory is retained for 30 days.\\n%sql\\n-- Review history by Delta table file path\\nDESCRIBE  HISTORY delta.`/ml/loan_by_state `;\\n-- Review history by Delta table defined in the metastore as `loan_by_state`\\nDESCRIBE  HISTORY loan_by_state ;\\n-- Review last 5 operations by Delta table defined in the metastore as \\n`loan_by_state`\\nDESCRIBE  HISTORY loan_by_state  LIMIT 5;\\n%python\\nfrom delta.tables  import *\\ndeltaTable  = DeltaTable .forPath(spark, “/ml/loan_by_state ”)\\n# get the full history of the table\\nfullHistoryDF  = deltaTable .history()\\n# get the last 5 operations\\nlast5OperationsDF  = deltaTable .history(5)\\n36 | Chapter 1: Basic Operations on Delta Lakes%scala\\nimport io.delta.tables._\\nval deltaTable  = DeltaTable .forPath(spark, pathToTable )\\n// get the full history of the table\\nval fullHistoryDF  = deltaTable .history()\\n// get the last 5 operations\\nval lastOperationDF  = deltaTable .history(5)\\nThe following is a screenshot of the full history of the Delta table stored in ` /ml/\\nloan_by_state ` from this notebook.\\nFigure 1-19. Describe the history of a Delta table\\nNote:\\n•Some of the columns may be nulls because the corresponding information may\\nnot be available in your environment.\\n•Columns added in the future will always be added after the last column.\\n•For the most current reference, refer to Delta Lake history schema  and operation\\nmetric keys .\\nWe will dive further into the application of table history to time travel in Chapter 3:\\nTime Travel with Delta.\\nVacuum History\\nOver time, more files will accumulate into your Delta table; many of the older files\\nmay no longer be needed because they represent previously overwritten versions of\\ndata (e.g. UPDATE , DELETE , etc.). Therefore, you can remove files no longer referenced\\nby a Delta table and are older than the retention threshold by running the VACUUM\\ncommand on the table. Important note, VACUUM  is not triggered automatically. When\\nit is triggered, the default retention threshold for the files is 7 days, i.e. no longer ref‐\\nerenced files older than 7 days will be removed.\\nTable Utilities | 37%sql\\n-- vacuum files in a path-based table by default retention threshold\\nVACUUM delta.\\'/data/events\\'\\n-- vacuum files by metastore defined table by default retention threshold\\nVACUUM eventsTable\\n-- vacuum files by metastore defined table that are no longer required older \\nthan 100 hours old\\nVACUUM eventsTable   RETAIN 100 hours\\n-- dry run: get the list of files to be deleted\\nVACUUM eventsTable  DRY RUN\\n%python\\nfrom delta.tables  import *\\n# vacuum files in path-based tables\\ndeltaTable  = DeltaTable .forPath(spark, pathToTable )\\n# vacuum files in metastore-based tables\\ndeltaTable  = DeltaTable .forName(spark, tableName )\\n# vacuum files in path-based table by default retention threshold\\ndeltaTable .vacuum()\\n# vacuum files not required by versions more than 100 hours old\\ndeltaTable .vacuum(100)\\n%scala\\nimport io.delta.tables._\\n# vacuum files in path-based tables\\nval deltaTable  = DeltaTable .forPath(spark, pathToTable )\\n# vacuum files in metastore -based tables\\nval deltaTable  = DeltaTable .forName(spark, tableName )\\n# vacuum files in path-based table by default retention  threshold\\ndeltaTable .vacuum()\\n# vacuum files not required  by versions  more than 100 hours old\\ndeltaTable .vacuum(100)\\nConfigure  Log and Data History\\nLog history.    How much history your Delta table retains is configurable per table\\nusing the config spark.databricks.delta.logRetentionDuration , defaulting to 30\\ndays. The Delta Transaction Log is cleaned up during new commits to the table if the\\nlogs have passed the set retention period. This is done so that the Delta Log does not\\n38 | Chapter 1: Basic Operations on Delta Lakesgrow indefinitely, but the Delta Log only contains information about what files are in\\nthe table.\\nIn conjunction with the Log retention period, there are also the data files to consider.\\nEvery time data is changed in a Delta table, there are old files marked for deletion and\\nnew files added. While the Delta Log is cleaned up automatically on write, data files\\nmust be deleted explicitly with a call to the vacuum API. VACUUM  can be quite slow on\\ncloud object storage but needs very little resources, so it makes sense to schedule this\\nseparately to run on some cadence, e.g. weekly.\\nData history.    The config spark.databricks.delta.deletedFileRetentionDuration\\ncontrols how long ago the files were marked for deletion before they are deleted by\\nVACUUM . The default here is 7 days, as an application must actively call the API for the\\nfiles to be deleted.\\nParallel deletion of files  during vacuum.    When using V ACUUM, to configure Spark to\\ndelete files in parallel (based on the number of shuffle partitions) set the session con‐\\nfiguration \" spark.databricks.delta.vacuum.parallelDelete.enabled \" to \"true “.\\nWarning\\nWe do not recommend that you set a retention interval shorter than 7 days because\\nold snapshots and uncommitted files can still be in use by concurrent readers or writ‐\\ners to the table. If vacuum cleans up active files, concurrent readers can fail or, worse,\\ntables can be corrupted when vacuum deletes files that have not yet been committed.\\nDelta Lake has a safety check to prevent you from running a dangerous vacuum com‐\\nmand. If you are certain that there are no operations being performed on this table\\nthat take longer than the retention interval you plan to specify, you can turn off this\\nsafety check by setting the Apache Spark configuration property spark.data\\nbricks.delta.retentionDurationCheck.enabled  to false . Y ou must choose an\\ninterval that is longer than the longest-running concurrent transaction and the\\nlongest period that any stream can lag behind the most recent update to the table.\\nRetrieve Delta table details\\nDESCRIBE DETAIL  functionality allows you to review the table metadata including\\n(but not limited to) table size, schema, partition columns, and other metrics on file\\nand file sizes. Y ou can use the table name or file path to specify the Delta table as\\ndemonstrated below.\\n%sql\\n-- Describe detail using Delta file path\\nDESCRIBE  DETAIL delta.`/ml/loan_by_state `;\\nTable Utilities | 39-- Describe detail using metastore-defined Delta table\\nDESCRIBE  DETAIL loan_by_state ;\\nFigure 1-20. Describe Delta table details\\nBelow is a transposed view of these values.\\nColumn name value\\nformat delta\\nid 18ec29e5-7837-4c5f-a2ca-f05892b6298b\\nname loan_by_state\\ndescription null\\nlocation dbfs://ml/loan_by_state\\ncreatedAt 2021-02-03T05:17:00.146+0000\\nlastModified 2021-02-03T05:26:37.000+0000\\npartitionColumns []\\nnumFiles 292\\nsizeInBytes 308386\\nproperties {\"delta.checkpoint.writeStatsAsStruct”: “true\"}\\nminReaderVersion 1\\nminWriterVersion 2\\nFor the current schema of the columns described by DESCRIBE DETAIL , refer to Detail\\nschema .\\nNote, while the following metadata queries are not specific to Delta Lake, they are\\nhelpful when working with metastore-defined tables.\\nDESCRIBE TABLE\\nDescribe Table returns the basic metadata information of a table. The metadata infor‐\\nmation includes column name, column type and column comment. Optionally you\\ncan specify a partition spec or column name to return the metadata pertaining to a\\npartition or column respectively.\\nDisplay detailed information about the specified columns, including the column sta‐\\ntistics collected by the following command. Note, this command is applicable to only\\nmetastore-defined tables.\\n%sql\\nDESCRIBE  TABLE loan_by_state ;\\n40 | Chapter 1: Basic Operations on Delta Lakes \\n# result\\n+---------------+---------+-------+\\n|       col_name|data_type|comment|\\n+---------------+---------+-------+\\n|     addr_state|   string|       |\\n|          count|   bigint|       |\\n|      stream_no|      int|       |\\n|               |         |       |\\n| # Partitioning|         |       |\\n|Not partitioned|         |       |\\n+---------------+---------+-------+\\nAlong with the basic metadata information and partitioning, DESCRIBE TABLE EXTEN\\nDED returns detailed table information such as parent database, table name, location\\nof table, provider, table properties, etc.\\n%sql\\nDESCRIBE  TABLE EXTENDED  loan_by_state ;\\n+--------------------+--------------------+-------+\\n|            col_name|           data_type|comment|\\n+--------------------+--------------------+-------+\\n|          addr_state|              string|       |\\n|               count|              bigint|       |\\n|           stream_no|                 int|       |\\n|                    |                    |       |\\n|      # Partitioning|                    |       |\\n|     Not partitioned|                    |       |\\n|                    |                    |       |\\n|# Detailed Table ...|                    |       |\\n|                Name|denny_db.loan_by_...|       |\\n|            Location|dbfs:/ml/loan_by_...|       |\\n|            Provider|               delta|       |\\n|    Table Properties|[delta.checkpoint...|       |\\n+--------------------+--------------------+-------+\\nFor the current schema of the columns described by DESCRIBE TABLE, refer to\\nSpark SQL Guide > DESCRIBE TABLE .\\nGenerate a manifest file\\nTo allow non-Spark systems to query Delta Lake without querying the transaction\\nlog, you can create a manifest file that those systems will query. To learn more about\\nhow to configure systems like Presto and Athena to read a Delta table, jump to Chap‐\\nter 9: Integrating with other engines.\\nThe following code generates the manifest file itself.\\n%sql\\n-- Generate manifest\\nGENERATE  symlink_format_manifest  FOR TABLE delta.`<path-to-delta-table>`\\nTable Utilities | 41%python\\n# Generate manifest\\ndeltaTable  = DeltaTable .forPath(<path-to-delta-table>)\\ndeltaTable .generate (\"symlink_format_manifest\" )\\n%scala\\n// Generate manifest\\nval deltaTable  = DeltaTable .forPath(<path-to-delta-table>)\\ndeltaTable .generate (\"symlink_format_manifest\" )\\nThe preceding command creates the following file if the <path-to-delta-table> is\\ndefined as /ml/loan-by-state .\\n$/ml/loan-by-state.delta/_symlink_format_manifest\\n          \\nConvert a Parquet table to a Delta table\\nConverting an existing Parquet table to a Delta table in-place is straightforward. This\\ncommand lists all the files in the directory, creates a Delta Lake transaction log that\\ntracks these files, and automatically infers the data schema by reading the footers of\\nall Parquet files. If your data is partitioned, you must specify the schema of the parti‐\\ntion columns as a DDL-formatted string (that is, <column-name1> <type>,\\n<column-name2> <type>,  ...).\\n%sql\\n-- Convert non partitioned parquet table at path \\'<path-to-table>\\'\\nCONVERT TO DELTA parquet.`<path-to-table>`\\n-- Convert partitioned Parquet table at path \\'<path-to-table>\\' and partitioned \\nby integer columns named \\'part\\' and \\'part2\\'\\nCONVERT TO DELTA parquet.`<path-to-table>` PARTITIONED  BY (part int, part2 int)\\n%python\\nfrom delta.tables  import *\\n# Convert non partitioned parquet table at path \\'<path-to-table>\\'\\ndeltaTable  = DeltaTable .convertToDelta (spark, \"parquet.`<path-to-table>`\" )\\n# Convert partitioned parquet table at path \\'<path-to-table>\\' and partitioned \\nby integer column named \\'part\\'\\npartitionedDeltaTable  = DeltaTable .convertToDelta (spark, \"parquet.`<path-to-\\ntable>`\" , \"part int\" )\\n%scala\\nimport io.delta.tables._\\n// Convert non partitioned Parquet table at path \\'<path-to-table>\\'\\nval deltaTable  = DeltaTable .convertToDelta (spark, \"parquet.`<path-to-table>`\" )\\n// Convert partitioned Parquet table at path \\'<path-to-table>\\' and partitioned \\nby integer columns named \\'part\\' and \\'part2\\'\\n42 | Chapter 1: Basic Operations on Delta Lakesval partitionedDeltaTable  = DeltaTable .convertToDelta (spark, \"parquet.`<path-to-\\ntable>`\" , \"part int, part2 int\" )\\nSome important notes:\\n•If a Parquet table was created by Structured Streaming, the listing of files can be\\navoided by using the _spark_metadata  sub-directory as the source of truth for\\nfiles contained in the table by setting the SQL configuration spark.data\\nbricks.delta.convert.useMetadataLog  to true .\\n•Any file not tracked by Delta Lake is invisible and can be deleted when you run a\\nvacuum. Y ou should avoid updating or appending data files during the conver‐\\nsion process. After the table is converted, make sure all writes go through Delta\\nLake.\\nConvert a Delta table to a Parquet table\\nIn case you need to convert a Delta table to parquet you can follow the following\\nsteps:\\n1.If you have performed Delta Lake operations that can change the data files (for\\nexample, delete or merge), run VACUUM  with a retention of 0 hours to delete all\\ndata files that do not belong to the latest version of the table.\\n2.Delete the _delta_log  directory in the table directory.\\nRestore a table version\\nY ou can restore your Delta table to a previous version. As noted in the previous\\nReview table history section, a Delta table internally maintains historic versions of the\\ntable that enable it to be restored to an earlier state.\\n%sql\\n-- restore version 9 of metastore defined Delta table\\nINSERT OVERWRITE  INTO loan_by_state\\nSELECT * FROM loan_by_state  VERSION AS OF 9\\nThe RESTORE  command is currently only available on Databricks\\nIf you’re using Databricks, you can also use the RESTORE  command to simplify this\\nprocess.\\nTable Utilities | 43%sql\\nRESTORE TABLE loan_by_state  TO VERSION AS OF 9\\nRESTORE TABLE delta.`/ml/loan_by_state /` TO TIMESTAMP  AS OF 9\\n%python\\nfrom delta.tables  import *\\n# path-based Delta tables\\ndeltaTable  = DeltaTable .forPath(spark, `/ml/loan_by_state/` )\\n# metastore-based tables\\ndeltaTable  = DeltaTable .forName(spark, `loan_by_state` )\\n# restore table to version 9\\ndeltaTable .restoreToVersion (9)\\n%scala\\nimport io.delta.tables._\\n// path-based Delta tables\\nval deltaTable  = DeltaTable .forPath(spark, `/ml/loan_by_state/` )\\n// metastore-based tables\\nval deltaTable  = DeltaTable .forName(spark, `loan_by_state` )\\n// restore table to version 9\\ndeltaTable .restoreToVersion (9)\\nClone your Delta Tables\\nTable cloning or creating copies of tables in a data lake or data warehouse has several\\npractical uses. But, given the expansive volume of data in tables in a data lake and the\\nrate of its growth, making physical copies of tables is an expensive operation. Using\\nCLONE  makes the process simpler and cost-effective with the help of table clones.\\nWhat are clones anyway?\\nClones are replicas of a source table at a given point in time. They have the same\\nmetadata as the source table: same schema, constraints, column descriptions, statis‐\\ntics, and partitioning. However, they behave as a separate table with a separate lineage\\nor history. Any changes made to clones only affect the clone and not the source. Any\\nchanges that happen to the source during or after the cloning process also do not get\\nreflected in the clone due to snapshot isolation. As of this writing, there are two types\\nof clones: shallow or deep.\\nThe CLONE  command is currently only available on Databricks\\n44 | Chapter 1: Basic Operations on Delta LakesShallow Clones.    A shallow  (also known as Zero-Copy) clone only duplicates the meta‐\\ndata of the table being cloned; the data files of the table itself are not copied. This type\\nof cloning does not create another physical copy of the data resulting in minimal\\nstorage costs. Shallow clones are inexpensive and can be extremely fast to create.\\nThese clones are not self-contained and  depend on the source from which they we re\\ncloned as the source of data . If the files in the source that the clone depends on are\\nremoved, for example with VACUUM , a shallow clone may become unusable. Therefore,\\nshallow clones are typically used for short-lived use cases such as testing and experi‐\\nmentation.\\nDeep Clones.    Shallow clones are great for short-lived use cases, but some scenarios\\nrequire a separate and independent copy of the table’s data. A deep clone makes a full\\ncopy of the metadata and data files of the table being cloned. In that sense, it is similar\\nin functionality to copying with a CTAS command ( CREATE TABLE.. AS… SELECT… ).\\nBut it is simpler to specify since it makes a faithful copy of the original table at the\\nspecified version and you don’t need to re-specify partitioning, constraints, and other\\ninformation as you have to do with CTAS. In addition, it is much faster , robust, and\\ncan work in an incremental  manner against failures.\\nWith deep clones, we copy additional metadata, such as your streaming application\\ntransactions and COPY INTO  transactions, so you can continue your ETL applications\\nexactly where it left off on a deep clone.\\nWhere do clones help?\\nThere are many scenarios where you need a copy of your datasets – for exploring,\\nsharing, or testing ML models or analytical queries. Below are some example use\\ncases.\\nTesting and experimentation with a production table.    When users need to test a new ver‐\\nsion of their data pipeline they often have to rely on sample test datasets that are not\\nrepresentative of all the data in their production environment. Data teams may also\\nwant to experiment with various indexing techniques to improve the performance of\\nqueries against massive tables. These experiments and tests cannot be carried out in a\\nproduction environment without risking production data processes and affecting\\nusers.\\nIt can take many hours or even days, to spin up copies of your production tables for a\\ntest or a development environment. Add to that, the extra storage costs for your\\ndevelopment environment to hold all the duplicated data – there is a large overhead\\nin setting a test environment reflective of the production data. With a shallow clone,\\nthis is trivial:\\nTable Utilities | 45%sql\\n-- Shallow clone table\\nCREATE TABLE delta.`/some/test/location ` SHALLOW CLONE prod.events\\n%python\\n# Shallow clone table\\nDeltaTable .forName(\"spark\", \"prod.events\" ).clone(\"/some/test/location\" , isShal\\nlow=True)\\n%scala\\n// Shallow clone table\\nDeltaTable .forName(\"spark\", \"prod.events\" ).clone(\"/some/test/location\" , isShal\\nlow=true)\\nAfter creating a shallow clone of your table in a matter of seconds, you can start run‐\\nning a copy of your pipeline to test out your new code, or try optimizing your table in\\ndifferent dimensions to see how you can improve your query performance, and much\\nmore. These changes will only affect your shallow clone, not your original table.\\nStaging major changes to a production table.    Sometimes, you may need to perform\\nsome major changes to your production table. These changes may consist of many\\nsteps, and you don’t want other users to see the changes which you’re making until\\nyou’re done with all of your work. A shallow clone can help you out here:\\n%sql\\n-- Create a shallow clone\\nCREATE TABLE temp.staged_changes  SHALLOW CLONE prod.events;\\n-- Test out deleting table from shallow clone table\\nDELETE FROM temp.staged_changes  WHERE event_id  is null;\\n-- Update shallow clone table\\nUPDATE temp.staged_changes  SET change_date  = current_date () WHERE change_date  \\nis null;\\n...\\n-- Perform your verifications\\nOnce you’re happy with the results, you have two options. If no other change has\\nbeen made to your source table, you can replace your source table with the clone. If\\nchanges have been made to your source table, you can merge the changes into your\\nsource table.\\n%sql\\n-- If no changes have been made to the source\\nREPLACE TABLE prod.events CLONE temp.staged_changes ;\\n-- If the source table has changed\\nMERGE INTO prod.events USING temp.staged_changes\\nON events.event_id  <=> staged_changes .event_id\\nWHEN MATCHED THEN UPDATE SET *;\\n46 | Chapter 1: Basic Operations on Delta Lakes-- Drop the staged table\\nDROP TABLE temp.staged_changes ;\\nMachine Learning result reproducibility.    Training machine learning models is typically\\nan iterative process. Throughout this process of optimizing the different parts of the\\nmodel, data scientists need to assess the accuracy of the model against a fixed dataset.\\nThis is hard to do in a system where the data is constantly being loaded or updated. A\\nsnapshot of the data used to train and test the model is required. This snapshot allows\\nthe results of the ML model to be reproducible for testing or model governance pur‐\\nposes. We recommend leveraging Time Travel to run multiple experiments across a\\nsnapshot; an example of this in action can be seen in Machine Learning Data Lineage\\nwith MLflow and Delta Lake . Once you’re happy with the results and would like to\\narchive the data for later retrieval, for example, next Black Friday, you can use deep\\nclones to simplify the archiving process. MLflow integrates really well with Delta\\nLake, and the auto logging feature ( mlflow.spark.autolog () ) will tell you, which\\nversion of the table was used to run a set of experiments.\\n# Run your ML workloads using Python and then\\nDeltaTable .forName(spark, \"feature_store\" ).cloneAtVersion (128, \"fea\\nture_store_bf2020\" )\\nData Migration.    A massive table may need to be moved to a new, dedicated bucket or\\nstorage system for performance, disaster recovery, or governance reasons. The origi‐\\nnal table will not receive new updates going forward and will be deactivated and\\nremoved at a future point in time. Deep clones make the copying of massive tables\\nmore robust and scalable.\\n%sql\\n-- Data migration using deep clone\\nCREATE TABLE delta.`zz://my-new-bucket/events` CLONE prod.events;\\nALTER TABLE prod.events SET LOCATION  \\'zz://my-new-bucket/events\\' ;\\nWith deep clones, since we copy your streaming application transactions and COPY\\nINTO transactions, you can continue your ETL applications from exactly where it\\nleft off after  this migration !\\nData Sharing.    It is common for users from different departments to look for data sets\\nthat they can use to enrich their analysis or models and you may want to share your\\ndata with them. But rather than setting up elaborate pipelines to move the data to yet\\nanother store, it is often easier and economical to create a copy of the relevant data\\nset for users to explore and test the data to see if it is a fit for their needs without\\naffecting your own production systems. Here deep clones significantly simplify this\\nprocess.\\n%sql\\n-- The following code can be scheduled to run at your convenience\\nTable Utilities | 47-- for data sharing\\nCREATE OR REPLACE TABLE data_science .events CLONE prod.events;\\nData Archiving.    For regulatory or archiving purposes all data in a table needs to be\\npreserved for a certain number of years, while the active table retains data for a few\\nmonths. If you want your data to be updated as soon as possible, but however you\\nhave a requirement to keep data for several years, storing this data in a single table\\nand performing time travel may become prohibitively expensive. In this case, archiv‐\\ning your data in a daily, weekly, or monthly manner is a better solution. The incre‐\\nmental cloning capability of deep clones will really help you here.\\n%sql\\n-- The following code can be scheduled to run at your convenience\\n-- for data archiving\\nCREATE OR REPLACE TABLE archive.events CLONE prod.events;\\nNote that this table will have an independent  history compared to the source table,\\ntherefore time travel queries on the source table and the clone may return different\\nresults based on your frequency of archiving.\\nFor more on clones\\nFor more information on clones:\\n•Delta CLONE Language Manual\\n•How to Easily Clone Y our Delta Lake Data Tables with Databricks\\n•Attack of the Delta Clones (Against Disaster Recovery Availability Complexity)\\n•We will have an Operational scenarios chapter for more examples\\nSummary\\nIn this chapter, we began by covering the three simple steps for you to get started on\\nDelta Lake: using --packages , building standalone applications via GitHub and\\nmaven, and using Databricks community edition. We provided the basic operations\\nof Delta Lake which look suspiciously similar to Spark basic operations - because they\\nare! We provided an intermediate-level primer on how the Delta Lake transaction log\\nworks and how it can provide ACID transactions to provide data reliability . Finally,\\nwe described some of the Delta Lake table utility commands.\\nBecause the Delta transaction log uses MVCC, there can be many files associated with\\ndifferent versions of the Delta table. Fortunately, the table utility commands simplify\\nthis process. But there are distinct feature advantages such as Delta time travel (or\\ndata versioning), which we cover in our next chapter, Chapter 3: Time Travel with\\nDelta .\\n48 | Chapter 1: Basic Operations on Delta LakesCHAPTER 2\\nTime Travel with Delta Lake\\nA Note for Early Release Readers\\nWith Early Release ebooks, you get books in their earliest form—the authors’ raw and\\nunedited content as they write—so you can take advantage of these technologies long\\nbefore the official release of these titles.\\nThis will be the 3rd chapter of the final book. Please note that the GitHub repo will be\\nmade active later on.\\nIf you have comments about how we might improve the content and/or examples in\\nthis book, or if you notice missing material within this chapter, please reach out to the\\neditor at gobrien@oreilly.com .\\nIntroduction\\nData engineering pipelines often go awry, especially when ingesting “dirty” data from\\nexternal systems. However, in a traditional data lake design, it is hard to undo updates\\nthat added objects into a table. In addition, it is challenging to audit data changes\\nwhich are critical, both in terms of data compliance as well as simple debugging, to\\nunderstand how data has changed over time. Other data workloads, such as machine\\nlearning training, require faithfully reproducing an old version of the data (e.g., to\\ncompare a new and old training algorithm on the same data). For example, if some\\nupstream pipeline modifies the source data, data scientists are often caught unaware\\nby such upstream data changes and hence struggle to reproduce their experiments.\\nAll of these issues created significant challenges for data engineers and data scientists\\nbefore Delta Lake, requiring them to design complex remediations to data pipeline\\nerrors or to haphazardly duplicate datasets. Through this chapter, we will explore the\\n49capabilities of Delta Lakes that allow data practitioners to go back in time and solve\\nfor these challenges. It is uncertain if time travel to the past is physically possible but\\ndata practitioners can time travel programmatically with Delta Lake. Delta Lake\\nallows automatic versioning of all data stored in the data lake and we can time travel\\nto any version. It also allows us to create a copy of an existing Delta table at a specific\\nversion using the clone  command to help with a few time travel use cases. To under‐\\nstand these functionalities and how time travel works, we first need to unpack the file\\nstructure of Delta tables and deep dive into each of the components.\\nUnder the hood of a Delta Table\\nIn chapter 2 , one of our sections was to focus on unpacking the transaction log. But\\nrecall that the transaction log is just one part of the Delta table. In this section, we will\\nfocus on the components that make up the Delta table.\\nThe Delta Directory\\nA Delta table is stored within a directory and is composed of the file types shown in\\nFigure 2-1.\\nFigure 2-1. Delta Directory structure\\nAs an example, we have a delta table named customer_data  located in the filePath:\\ndbfs:/user/hive/warehouse/deltaguide.db\\nTo visualize the delta file system, see below a snippet from our example. Y ou can also\\nfollow along with us by using the notebooks present here .\\nRoot of the Delta directory : This is the hive warehouse location where we created a\\ndelta table and using the command below, we can explore the contents within the\\ndelta directory. For the full output, please refer to  this location .\\n%fs ls dbfs:/user/hive/warehouse/deltaguide.db/customer_data\\nTable 2-1. Delta Root Directory\\npath name size\\n50 | Chapter 2: Time Travel with Delta Lakedbfs:/dbfs:/..filePath../customer_data/_delta_log/ _delta_log/ 0\\ndbfs:/dbfs:/..filePath../customer_data/part-00000-41…-\\nc001.snappy.parquetpart-00000-41...-c001.snappy.parquet 542929260\\ndbfs:/dbfs:/..filePath../customer_data/part-00000-45...-\\nc002.snappy.parquetpart-00000-45...-c002.snappy.parquet 461126371\\ndbfs:/dbfs:/..filePath../customer_data/part-00000-62...-\\nc000.snappy.parquetpart-00000-62...-c000.snappy.parquet 540468009\\ndbfs:/dbfs:/..filePath../customer_data/part-00001-40...-\\nc002.snappy.parquetpart-00001-40...-c002.snappy.parquet 541858629\\ndbfs:/dbfs:/..filePath../customer_data/part-00001-a7...-\\nc001.snappy.parquetpart-00001-a7...-c001.snappy.parquet 542859315\\ndbfs:/dbfs:/..filePath../customer_data/part-00001-c3...-\\nc000.snappy.parquetpart-00001-c3...-c000.snappy.parquet 541186721\\nDelta Logs Directory\\nWhen a user creates a Delta Lake table, that table’s transaction log is automatically\\ncreated in the _delta_log subdirectory. As he or she makes changes to that table, those\\nchanges are recorded as ordered, atomic commits in the transaction log. Each com‐\\nmit is written out as a JSON file, starting with 000000.json. Additional changes to the\\ntable generate subsequent JSON files in ascending numerical order so that the next\\ncommit is written out as 000001.json, the following as 000002.json, and so on.\\nFigure 2-2 shows the logical diagram of Delta Transaction Log Protocol.\\nFigure 2-2. Delta Transaction Log Protocol\\nThe state of a table at a given version is called a snapshot  and includes the following\\nproperties:\\n•Version of the Delta log protocol : This is required to correctly read or write the\\ntable\\n•Metadata of the table (e.g., the schema, a unique identifier, partition columns,\\nand other configuration properties)\\n•Set of files  present in the table, along with metadata about those files\\n•Set of tombstones  for files that were recently deleted\\nUnder the hood of a Delta Table | 51•Set of applications-specific  transactions that have been successfully committed\\nto the table\\nWe will explore each of the properties in the upcoming sections and to provide you a\\nmental mapping of how this actually looks from a file system perspective, here is a\\nsnippet.\\n%fs ls dbfs:/fil  ePath/customer_t2/_delta_log\\nTable 2-2. Delta Logs Directory\\npath name size\\ndbfs:/.../customer_t2/_delta_log/.s3-optimization-0 .s3-optimization-0 0\\ndbfs:/.../customer_t2/_delta_log/.s3-optimization-1 .s3-optimization-1 0\\ndbfs:/.../customer_t2/_delta_log/.s3-optimization-2 .s3-optimization-2 0\\ndbfs:/.../customer_t2/_delta_log/\\n00000000000000000010.checkpoint.parquet00000000000000000010.checkpoint.parquet 120950\\n:\\n:\\n::\\n:\\n::\\n:\\n:\\ndbfs:/.../customer_t2/_delta_log/\\n00000000000000000020.checkpoint.parquet00000000000000000020.checkpoint.parquet 94341\\ndbfs:/.../customer_t2/_delta_log/00000000000000000020.crc 00000000000000000020.crc 97\\ndbfs:/.../customer_t2/_delta_log/00000000000000000020.json 00000000000000000020.json 347291\\ndbfs:/.../customer_t2/_delta_log/00000000000000000021.crc 00000000000000000021.crc 95\\ndbfs:/.../customer_t2/_delta_log/00000000000000000021.json 00000000000000000021.json 41461\\ndbfs:/.../customer_t2/_delta_log/_last_checkpoint _last_checkpoint 26\\nAs previously noted, Delta Lake is an open-source storage layer that runs on top of\\nyour existing data lake and is fully compatible with Apache Spark APIs. Delta Lake\\nuses versioned Parquet files to store data in your cloud storage, enabling Delta Lake\\nto leverage the efficient compression and encoding schemes that are native to Par‐\\nquet. Apart from the versions, Delta Lake also stores a transaction log to keep track of\\nall the commits made to the table or blob store directory to provide ACID transac‐\\ntions.\\nY ou can use your favorite Apache Spark APIs to read and write data with Delta Lake.\\nAs noted in the previous chapter, to create a Delta table, write a DataFrame out in the\\ndelta  format.\\n%python\\n# Generate Spark DataFrame\\ndata = spark.range(0, 5)\\n# Write the table in parquet format\\ndata.write.format(\"parquet\" ).save(\"/table_pq\" )\\n52 | Chapter 2: Time Travel with Delta Lake# Write the table in delta format\\ndata.write.format(\"delta\").save(\"/table_delta\" )\\n%scala\\n// Generate Spark DataFrame\\nval data = spark.range(0, 5)\\n// Write the table in parquet format\\ndata.write.format(\"parquet\" ).save(\"/table_pq\" )\\n// Write the table in delta format\\ndata.write.format(\"delta\").save(\"/table_delta\" )\\nOnce you write the files in delta, you will notice that a number of files are created\\nunder a folder in the format recognized by Spark as a Parquet table and/or stored in a\\nmetastore (e.g. Hive, Glue, etc.).\\nNote, the files of a metastore-defined table are stored under the\\ndefault hive/warehouse  location though you have an option to\\nspecify a location of your choice at the time of writing delta files as\\nnoted in the previous example.\\nTo review the underlying file structure, run the following command from your\\n            -table>\\n          \\nThe following is the shell command for the parquet table previously generated.\\n%sh ls -R /dbfs/table_pq/\\n/dbfs/table_pq/:\\npart-00000-775edc03-7c05-4190-a964-fcdfc0428014-c000.snappy.parquet\\npart-00001-94940738-a4e6-4b4b-b30f-797a76b32087-c000.snappy.parquet\\npart-00003-0e5d49e6-4dd0-4746-8944-f64cba9df97c-c000.snappy.parquet\\npart-00004-bed772f9-1045-4e3a-8048-e179096e3c25-c000.snappy.parquet\\npart-00006-e28f9fa4-516b-4cf6-acdd-22d70e8841f2-c000.snappy.parquet\\npart-00007-5276d994-b9ca-4da8-b594-4bbfafbd7dcb-c000.snappy.parquet\\nThe following is the shell command for the Delta table previously generated.\\n%sh ls -R /dbfs/table_delta/\\n/dbfs/table_delta/:\\n_delta_log\\npart-00000-775edc03-7c05-4190-a964-fcdfc0428014-c000.snappy.parquet\\npart-00001-94940738-a4e6-4b4b-b30f-797a76b32087-c000.snappy.parquet\\npart-00003-0e5d49e6-4dd0-4746-8944-f64cba9df97c-c000.snappy.parquet\\npart-00004-bed772f9-1045-4e3a-8048-e179096e3c25-c000.snappy.parquet\\npart-00006-e28f9fa4-516b-4cf6-acdd-22d70e8841f2-c000.snappy.parquet\\nUnder the hood of a Delta Table | 53part-00007-5276d994-b9ca-4da8-b594-4bbfafbd7dcb-c000.snappy.parquet\\n/dbfs/table_delta/_delta_log:\\n00000000000000000000.crc\\n00000000000000000000.json\\nMajority of Data Practitioners are familiar with Parquet and its structure, let’s sim‐\\nplify the understanding of Delta by comparing it to Parquet.\\nWhat is the difference  between the Parquet and Delta tables?\\nThe only difference between the Parquet and Delta tables is the _delta_log  folder\\nwhich is the Delta transaction log (more on this in Chapter 2).\\nWhat is the same between the Parquet and Delta tables?\\nIn both Parquet and Delta tables, the data itself is in the form of snappy compressed\\nParquet part files, which is the common format of Spark tables.\\n•Apache Parquet  is a columnar storage format that is optimized for BI-type quer‐\\nies (group by, aggregates, joins, etc.). Because of this, it is the default storage for‐\\nmat for Apache Spark when it writes its data files to storage.\\n•Snappy compression  was developed by Google to provide high compression/\\ndecompression speeds with reasonable compression. By default, Spark tables are\\nsnappy compressed Parquet files.\\n•By default, the reference implementation of Delta lake stores data files in directo‐\\nries named after partition values for data in that file (i.e. part1=value1/\\npart2=value2/... ) that is why we see part files and a suffix in the name as a\\nglobally unique identifier  to ensure uniqueness of each file. This is especially\\nimportant because each directory in cloud object stores are not actual folders but\\nlogical representations of folders. For more information on the implications, read\\nmore at How to list and delete files faster in Databricks .\\nThe files  of a Delta table\\nny time data is modified in a Delta table, new files are created as a Version which is a\\nsnapshot of the delta table at that specific time and is a result of a set of actions that\\nwere performed by the user.\\nVersion 0: Table creation\\nLet’s start by looking at the Delta table created in the previous section by running the\\nfollowing commands to review the path  column of the add metadata.\\n%python\\n# Read first transaction\\nj0 = spark.read.json(\"/table_delta/_delta_log/00000000000000000000.json\" )\\n54 | Chapter 2: Time Travel with Delta Lake# Review Add Information\\nj0.select(\"add.path\" ).where(\"add is not null\" ).show(20, False)\\n%scala\\n// Read first transaction\\nval j0 = spark.read.json(\"/table_delta/_delta_log/00000000000000000000.json\" )\\n// Review Add Information\\nj0.select(\"add.path\" ).where(\"add is not null\" ).show(20, False)\\nMetadata in the transaction log include all of the files that make up the table version\\n(version 0). The output of the preceding commands is:\\n+-------------------------------------------------------------------+\\n|path                                                               |\\n+-------------------------------------------------------------------+\\n|part-00000-775edc03-7c05-4190-a964-fcdfc0428014-c000.snappy.parquet|\\n|part-00001-94940738-a4e6-4b4b-b30f-797a76b32087-c000.snappy.parquet|\\n|part-00003-0e5d49e6-4dd0-4746-8944-f64cba9df97c-c000.snappy.parquet|\\n|part-00004-bed772f9-1045-4e3a-8048-e179096e3c25-c000.snappy.parquet|\\n|part-00006-e28f9fa4-516b-4cf6-acdd-22d70e8841f2-c000.snappy.parquet|\\n|part-00007-5276d994-b9ca-4da8-b594-4bbfafbd7dcb-c000.snappy.parquet|\\n+-------------------------------------------------------------------+\\nNotice how this matches the file listing you see when you run the ls command.\\nWhile this file listing is pretty fast for small amounts of data, as noted in  Chapter 2,\\nlistFrom  can be quite slow on cloud object storage especially for petabyte-scale data\\nlakes.\\nWhen reviewing the table history (see Table 3-2), let’s focus on the operationMetrics\\nfor version 0 (table creation).\\n%sql\\n-- Describe table history using file path\\nDESCRIBE  HISTORY delta.`/table_delta `\\nTable 3-x transposes the values of the preceding figure.\\nColumn Name Value\\nversion 0\\ntimestamp userID\\nuserName vini[dot]jaiswal[at]databricks.com\\noperation WRITE\\noperationParameters {\"mode”: “ErrorIfExists”, “partitionBy”: “[]\"}\\njob null\\nnotebook {\"notebookId”: “9342327\"}\\nclusterId  \\nreadVersion null\\nisolationLevel WriteSerializable\\nUnder the hood of a Delta Table | 55isBlindAppend true\\noperationMetrics {\"numFiles”: “6”, “numOutputBytes”: “2713”, “numOutputRows”: “5\"}\\nuserMetadata null\\nOf particular interest in Table 3-x is the operationMetrics  column for table version 0.\\n{\\n\"numFiles\": \"6\", \\n\"numOutputBytes\": \"2713\", \\n\"numOutputRows\": \"5\"\\n}\\nNotice how numFiles  corresponds to the six files listed in the previous add.path\\nquery as well as the file listing ( ls).\\nVersion 1: Appending data\\nWhat happens when we add new data to this table? The following command will add\\n4 new rows to our table.\\n%python\\n# Add 4 new rows of data to our Delta table\\ndata = spark.range(6, 10)\\ndata.write.format(\"delta\").mode(\"append\" ).save(\"/table_delta\" )\\n%scala\\n// Add 4 new rows of data to our Delta table\\nval data = spark.range(6, 10)\\ndata.write.format(\"delta\").mode(\"append\" ).save(\"/table_delta\" )\\nWe’re now going to go over how to review data, file system, added files, etc.\\nReview data.    Y ou can confirm there are a total number of 9 rows in the table by run‐\\nning the following command.\\n%python\\nspark.read.format(\"delta\").load(\"/table_delta\" ).count()\\n%scala\\nspark.read.format(\"delta\").load(\"/table_delta\" ).count()\\nReview file system.    But what does the underlying file system look like? When we re-\\nrun our file listing, as expected, there are more files and a new JSON and CRC file\\nwithin _delta_log  corresponding to a new version of the table.\\n%sh ls -R /dbfs/table_delta/\\n/dbfs/table_delta/:\\n_delta_log\\npart-00000-0267038a-f818-4823-8261-364fd7401501-c000.snappy.parquet\\npart-00000-775edc03-7c05-4190-a964-fcdfc0428014-c000.snappy.parquet\\npart-00001-94940738-a4e6-4b4b-b30f-797a76b32087-c000.snappy.parquet\\n56 | Chapter 2: Time Travel with Delta Lakepart-00001-cd3a1a49-0a0a-4284-8a1b-283d83590dff-c000.snappy.parquet\\npart-00003-0e5d49e6-4dd0-4746-8944-f64cba9df97c-c000.snappy.parquet\\npart-00003-56bc7589-6266-4ce3-9515-62caf9af9109-c000.snappy.parquet\\npart-00004-bed772f9-1045-4e3a-8048-e179096e3c25-c000.snappy.parquet\\npart-00005-3d298fe6-8795-4558-92c7-70e0520a3d47-c000.snappy.parquet\\npart-00006-e28f9fa4-516b-4cf6-acdd-22d70e8841f2-c000.snappy.parquet\\npart-00007-5276d994-b9ca-4da8-b594-4bbfafbd7dcb-c000.snappy.parquet\\npart-00007-60de290b-7147-4ec4-b535-ca9abf4ce2d1-c000.snappy.parquet\\n/dbfs/table_delta/_delta_log:\\n00000000000000000000.crc\\n00000000000000000000.json\\n00000000000000000001.crc\\n00000000000000000001.json\\nReview added files .    Let’s follow up by looking at the Delta table version 1 by running\\nthe following commands to review the path  column of the add metadata.\\n%python\\n# Read version 1\\nj1 = spark.read.json(\"/table_delta/_delta_log/00000000000000000001.json\" )\\n# Review Add Information\\nj1.select(\"add.path\" ).where(\"add is not null\" ).show(20, False)\\n%scala\\n// Read version 1\\nj1 = spark.read.json(\"/table_delta/_delta_log/00000000000000000001.json\" )\\n// Review Add Information\\nj1.select(\"add.path\" ).where(\"add is not null\" ).show(20, False)\\nMetadata in the transaction log include all of the files that make up table version 1.\\nThe output of the preceding commands is:\\n+-------------------------------------------------------------------+\\n|path                                                               |\\n+-------------------------------------------------------------------+\\n|part-00000-0267038a-f818-4823-8261-364fd7401501-c000.snappy.parquet|\\n|part-00001-cd3a1a49-0a0a-4284-8a1b-283d83590dff-c000.snappy.parquet|\\n|part-00003-56bc7589-6266-4ce3-9515-62caf9af9109-c000.snappy.parquet|\\n|part-00005-3d298fe6-8795-4558-92c7-70e0520a3d47-c000.snappy.parquet|\\n|part-00007-60de290b-7147-4ec4-b535-ca9abf4ce2d1-c000.snappy.parquet|\\n+-------------------------------------------------------------------+\\nInterestingly, this output notes that only 5 files were needed to hold the 4 new rows of\\ndata; recall there were 6 files for the initial 5 rows of data.\\nReview table history.    This can be confirmed by reviewing the table history, let’s focus\\non the operationMetrics for version 1 (table creation).\\nUnder the hood of a Delta Table | 57%sql\\n-- Describe table history using file path\\nDESCRIBE  HISTORY delta.`/table_delta `\\nBelow is the value of the operationMetrics  column of version 1 which confirmed in\\nfact there are only 5 files with the number of output rows being 4.\\n{\\n\"numFiles\": \"5\", \\n\"numOutputBytes\": \"2232\", \\n\"numOutputRows\": \"4\"\\n}\\nNotice how numFiles  corresponds to the five files listed in the previous add.path\\nquery as well as the file listing ( ls).\\nQuery added files .    Y ou can further validate this by running the following query\\n%sql\\n-- Run this statement first as Delta will do a format check\\nSET spark.databricks .delta.formatCheck .enabled=false\\n%python\\ndelta_path  = \"/table_delta/\"\\n# Files listed in add.path metadata\\nfiles = [delta_path + \"part-00000-0267038a-f818-4823-8261-364fd7401501-\\nc000.snappy.parquet\",\\n         delta_path + \"part-00001-cd3a1a49-0a0a-4284-8a1b-283d83590dff-\\nc000.snappy.parquet\",\\n         delta_path + \"part-00003-56bc7589-6266-4ce3-9515-62caf9af9109-\\nc000.snappy.parquet\",\\n         delta_path + \"part-00005-3d298fe6-8795-4558-92c7-70e0520a3d47-\\nc000.snappy.parquet\",\\n         delta_path + \"part-00007-60de290b-7147-4ec4-b535-ca9abf4ce2d1-\\nc000.snappy.parquet\"]\\n# Show values stored in these files\\nspark.read.format(\"parquet\").load(files).show()\\n# Results\\n+---+\\n| id|\\n+---+\\n|  8|\\n|  6|\\n|  7|\\n|  9|\\n+---+\\nAs noted, these five files correspond to the four rows (id values 6, 7, 8, and 9) added\\nto our Delta table.\\n58 | Chapter 2: Time Travel with Delta LakeVersion 2: Deleting data\\nWhat happens when we remove data from this table? The following command will\\nDELETE  some of the values from our Delta table. Note, we will dive deeper into data\\nmodifications such as delete, update, and merge in Chapter 5: Data Modifications in\\nDelta tables. But this example is an interesting showcase of what is happening in the\\nfile system.\\n%sql\\n-- Delete from Delta table where id <= 2\\nDELETE FROM delta.`/table_delta` WHERE id <= 2\\nReview data.    Y ou can confirm there are a total number of 6 rows (the 3 values of 0, 1,\\nand 2 were removed) in the table by running the following command.\\n%python\\nspark.read.format(\"delta\").load(\"/table_delta\").count()\\n%scala\\nspark.read.format(\"delta\").load(\"/table_delta\").count()\\nReview file system.    But what does the underlying file system look like? Let’s re-run\\nour file listing and as expected, there are more files and a new JSON and CRC file\\nwithin _delta_log  corresponding to a new version of the table.\\n%sh ls -R /dbfs/table_delta/\\n/dbfs/table_delta/:\\n_delta_log\\npart-00000-0267038a-f818-4823-8261-364fd7401501-c000.snappy.parquet\\npart-00000-74e6c7ea-7321-44f0-9fb6-55257677cb1f-c000.snappy.parquet\\npart-00000-775edc03-7c05-4190-a964-fcdfc0428014-c000.snappy.parquet\\npart-00001-94940738-a4e6-4b4b-b30f-797a76b32087-c000.snappy.parquet\\npart-00001-cd3a1a49-0a0a-4284-8a1b-283d83590dff-c000.snappy.parquet\\npart-00003-0e5d49e6-4dd0-4746-8944-f64cba9df97c-c000.snappy.parquet\\npart-00003-56bc7589-6266-4ce3-9515-62caf9af9109-c000.snappy.parquet\\npart-00004-bed772f9-1045-4e3a-8048-e179096e3c25-c000.snappy.parquet\\npart-00005-3d298fe6-8795-4558-92c7-70e0520a3d47-c000.snappy.parquet\\npart-00006-e28f9fa4-516b-4cf6-acdd-22d70e8841f2-c000.snappy.parquet\\npart-00007-5276d994-b9ca-4da8-b594-4bbfafbd7dcb-c000.snappy.parquet\\npart-00007-60de290b-7147-4ec4-b535-ca9abf4ce2d1-c000.snappy.parquet\\n/dbfs/table_delta/_delta_log:\\n00000000000000000000.crc\\n00000000000000000000.json\\n00000000000000000001.crc\\n00000000000000000001.json\\n00000000000000000002.crc\\n00000000000000000002.json\\nUnder the hood of a Delta Table | 59As expected, there is a new transaction log file ( 000...00002.json ) but now there are\\n12 files in our Delta table folder after we deleted three rows (previously there were 11\\nfiles). Let’s review Delta table version 2 by running the following commands to review\\nthe path  column of the add and remove  metadata.\\nReview removed files .    There were three files removed  from our table; the following\\nquery reads the version 2 transaction log and specifically extracts the path of removed\\nfiles.\\n%python\\n# Remove information\\nj2.select(\"remove.path\" ).where(\"remove is not null\" ).show(20, False)\\n%scala\\n// Remove information\\nj2.select(\"remove.path\" ).where(\"remove is not null\" ).show(20, False)\\n# Result\\n+-------------------------------------------------------------------+\\n|path                                                               |\\n+-------------------------------------------------------------------+\\n|part-00004-bed772f9-1045-4e3a-8048-e179096e3c25-c000.snappy.parquet|\\n|part-00003-0e5d49e6-4dd0-4746-8944-f64cba9df97c-c000.snappy.parquet|\\n|part-00001-94940738-a4e6-4b4b-b30f-797a76b32087-c000.snappy.parquet|\\n+-------------------------------------------------------------------+\\nIt is important to note that in the transaction log the deletion is a tombstone, i.e. we\\nhave not deleted the files. That is, these files are identified as removed  so that when\\nyou query the table, Delta will not include the remove  files.\\nReview table history.    This can be confirmed by reviewing the table history, let’s focus\\non the operationMetrics for version 2 (table creation).\\n%sql\\n-- Describe table history using file path\\nDESCRIBE  HISTORY delta.`/table_delta `\\nBelow is the value of the operationMetrics  column of version 2 which confirmed in\\nfact that 3 files were removed, 1 file was added, and 3 rows were deleted.\\n{\\n\"numRemovedFiles\": \"3\", \\n\"numDeletedRows\": \"3\", \\n\"numAddedFiles\": \"1\", \\n\"numCopiedRows\": \"0\"\\n}\\nQuery removed files .    As stated earlier, the files may be removed in the transaction log,\\nbut they are not removed from the file system. The Delta data files are created in this\\nfashion to correspond to the concept of multi-version concurrency control (MVCC).\\n60 | Chapter 2: Time Travel with Delta LakeFor example, as seen in Figure 3-x, if user 1 is querying version 1 of the table while\\nuser 2 is deleting values (0, 1, 2) that creates table version 2. if the data  files that rep‐\\nresent version 1 (that contained the values 0, 1, and 2), user 1’s query would irrevoca‐\\nbly fail.\\nFigure 2-3. Concurrent queries against different  versions of the data\\nThis is especially important for long-running queries on data lakes where the queries\\nand operations may take a long time. It is important to note that by default user 1 had\\nexecuted the read query before  the version 2 of the table was committed. If at a later\\ntime, as seen in Figure 3-x, another user (user 3) queries the same table, then by\\ndefault when they read the data, the three files representing values {0, 1, 2} would not\\nbe included and they would only be returned six rows.\\nFigure 2-4. Another user querying the data after  the delete has completed\\nUnder the hood of a Delta Table | 61To further validate the data is still there, run the following query; the three files\\n(files ) correspond to the three removed  files in the previous section.\\n%sql\\n-- Run this statement first as Delta will do a format check\\nSET spark.databricks .delta.formatCheck .enabled=false\\n%python\\ndelta_path  = \"/table_delta/\"\\n# Files listed in add.path metadata\\nfiles = [delta_path + \"part-00004-bed772f9-1045-4e3a-8048-e179096e3c25-\\nc000.snappy.parquet\",\\n         delta_path + \"part-00003-0e5d49e6-4dd0-4746-8944-f64cba9df97c-\\nc000.snappy.parquet\",\\n         delta_path + \"part-00001-94940738-a4e6-4b4b-b30f-797a76b32087-\\nc000.snappy.parquet\"]\\n# Show values stored in these files\\nspark.read.format(\"parquet\").load(files).show()\\n# Results\\n+---+\\n| id|\\n+---+\\n|  2|\\n|  1|\\n|  0|\\n+---+\\nReview added files .    Even though we had deleted three rows, we’ve also added a file in\\nthe file system as recorded in the transaction log.\\n%python\\n# Read version 2\\nj2 = spark.read.json(\"/table_delta/_delta_log/00000000000000000002.json\" )\\n># Review Add Information\\nj2.select(\"add.path\" ).where(\"add is not null\" ).show(20, False)\\n%scala\\n// Read version 2\\nj2 = spark.read.json(\"/table_delta/_delta_log/00000000000000000002.json\" )\\n// Review Add Information\\nj2.select(\"add.path\" ).where(\"add is not null\" ).show(20, False)\\nTo reiterate, even though we deleted three rows, as noted in Add Information  there is\\nan additional file!\\n+-------------------------------------------------------------------+\\n|path                                                               |\\n+-------------------------------------------------------------------+\\n62 | Chapter 2: Time Travel with Delta Lake|part-00000-74e6c7ea-7321-44f0-9fb6-55257677cb1f-c000.snappy.parquet|\\n+-------------------------------------------------------------------+\\nIn fact, there is nothing in this file if you were to read it using the query below as the\\ndescription of this column is apt.\\n%python\\n# Read the add file from version 2\\nspark.read.format(\"parquet\" ).load(\"/table_delta/\\npart-00000-74e6c7ea-7321-44f0-9fb6-55257677cb1f-c000.snappy.parquet\" ).show()\\n%scala\\n// Read the add file from version 2\\nspark.read.format(\"parquet\" ).load(\"/table_delta/\\npart-00000-74e6c7ea-7321-44f0-9fb6-55257677cb1f-c000.snappy.parquet\" ).show()\\n# Result\\n+---+\\n| id|\\n+---+\\n+---+\\nDuring our delete operation, no new rows or files were added to the table as was\\nexpected for this small example.\\nWhy are files  added during a delete?.    For larger datasets, it is common that not all the\\nrows will be removed from a file.\\nUnder the hood of a Delta Table | 63Figure 2-5. Deleting data can result in adding files\\nExtending the current scenario, let’s say that the values {3-8}  are stored in 1.parquet\\nand you run a DELETE  statement to remove the values id = 4 . In this example:\\n•Within the transaction log, the 1.parquet  file is added to the remove  column.\\n•But if we stop here, we would be removing not only {4} but also {3,5-8} .\\n•Therefore, a new file will be created - in this case the 2.parquet  file - that con‐\\ntains the values that were not deleted {3,5-8} .\\nTo demonstrate this with this scenario, let’s compact our current Delta table. If you’re\\nrunning this on Databricks, you can use the OPTIMIZE  command; more on this com‐\\nmand in Chapter 12: Performance Tuning.\\n%sql\\n-- Optimize our Delta table\\nOPTIMIZE  delta.`/table_delta /`\\nThis command is used to tune the performance of your Delta tables and it also has\\nthe benefit of compacting your files. In this case, this means that instead of 11 sepa‐\\nrate files, representing our 9 values, now we only have one file. The optimize opera‐\\ntion itself is the third operation per the table history.\\n64 | Chapter 2: Time Travel with Delta Lake%sql\\n-- Table history\\nDESCRIBE  HISTORY delta.`/table_delta /`\\n-- Abridged results\\n+-------+---------+----------------------------+\\n|version|operation|operationMetrics            |\\n+-------+---------+----------------------------+\\n|3      |OPTIMIZE |[numRemovedFiles -> 9,  ... |\\n|2      |DELETE   |[numRemovedFiles -> 3,  ... |\\n|1      |WRITE    |[numFiles -> 5,  ...        |\\n|0      |WRITE    |[numFiles -> 6,  ...        |\\n+-------+---------+----------------------------+\\nThe results from running the OPTIMIZE  command can be seen below (which is also\\nrecorded in the operationMetrics  column within table history.\\npath: /table_delta/\\nMetrics:\\n {\\n\"numFilesAdded\": 1, \\n\"numFilesRemoved\": 9, \\n\"filesAdded\": {\"min\": 507, \"max\": 507, \"avg\": 507, \"totalFiles\": 1, \"total-\\nSize\": 507},\\n \"filesRemoved\": {\"min\": 308, \"max\": 481, \"avg\": 423, \"totalFiles\": 9, \"total-\\nSize\": 3810},\\n \"partitionsOptimized\": 0, \\n\"zOrderStats\": null, \\n\"numBatches\": 1\\n}\\nThe two metrics most interesting for this scenario are numFilesAdded  and numFiles\\nRemoved  which denote that now there is only 1 file that contains all the values and 9\\nfiles were removed.\\nThe following query to validate the files that were removed.\\n%python\\n# Read version 3\\nj3 = spark.read.json(\"/table_delta/_delta_log/00000000000000000003.json\" )\\n# Remove Information\\nj3.select(\"remove.path\" ).where(\"add is not null\" ).show(20, False)\\n%scala\\n// Read version 3\\nval j3 = spark.read.json(\"/table_delta/_delta_log/00000000000000000003.json\" )\\n// Remove Information\\nj3.select(\"remove.path\" ).where(\"add is not null\" ).show(20, False)\\n-- Results\\n+-------------------------------------------------------------------+\\nUnder the hood of a Delta Table | 65|path                                                               |\\n+-------------------------------------------------------------------+\\n|part-00000-775edc03-7c05-4190-a964-fcdfc0428014-c000.snappy.parquet|\\n|part-00000-0267038a-f818-4823-8261-364fd7401501-c000.snappy.parquet|\\n|part-00000-74e6c7ea-7321-44f0-9fb6-55257677cb1f-c000.snappy.parquet|\\n|part-00006-e28f9fa4-516b-4cf6-acdd-22d70e8841f2-c000.snappy.parquet|\\n|part-00007-5276d994-b9ca-4da8-b594-4bbfafbd7dcb-c000.snappy.parquet|\\n|part-00001-cd3a1a49-0a0a-4284-8a1b-283d83590dff-c000.snappy.parquet|\\n|part-00003-56bc7589-6266-4ce3-9515-62caf9af9109-c000.snappy.parquet|\\n|part-00005-3d298fe6-8795-4558-92c7-70e0520a3d47-c000.snappy.parquet|\\n|part-00007-60de290b-7147-4ec4-b535-ca9abf4ce2d1-c000.snappy.parquet|\\n+-------------------------------------------------------------------+\\nThe following query allows you to validate the files that were added.\\n%python\\n# Add Information\\nj3.select(\"add.path\" ).where(\"add is not null\" ).show(20, False)\\n%scala\\n// Add Information\\nj3.select(\"add.path\" ).where(\"add is not null\" ).show(20, False)\\n-- Results\\n+-------------------------------------------------------------------+\\n|path                                                               |\\n+-------------------------------------------------------------------+\\n|part-00000-f661f932-f54f-4b38-8ed3-51f7770d3e55-c000.snappy.parquet|\\n+-------------------------------------------------------------------+\\nY ou can query the file and validate that it contains all the expected values {3-8} .\\n%sql\\n-- Temporarily disable the Delta format check\\nSET spark.databricks .delta.formatCheck .enabled=false\\n%python / %scala\\n# Python or Scala\\nspark.read.parquet(\"/table_delta/part-00000-f661f932-\\nf54f-4b38-8ed3-51f7770d3e55-c000.snappy.parquet\" ).show()\\n-- Results\\n+---+\\n| id|\\n+---+\\n|  4|\\n|  8|\\n|  3|\\n|  6|\\n|  9|\\n|  7|\\n+---+\\nSo now that all of the data is in one file, what happens when we delete a value in our\\ntable?\\n66 | Chapter 2: Time Travel with Delta Lake%sql\\nDELETE FROM delta.`/tmp/delta/` WHERE id = 4\\nAs we know this is version 4 of our table history, let’s review the transaction log using\\nthe following query.\\n%python\\n# Read version 4\\nj4 = spark.read.json(\"/table_delta/_delta_log/00000000000000000004.json\" )\\n# Remove Information\\nj4.select(\"remove.path\" ).where(\"add is not null\" ).show(20, False)\\n%scala\\n// Read version 4\\nval j4 = spark.read.json(\"/table_delta/_delta_log/00000000000000000004.json\" )\\n// Remove Information\\nj4.select(\"remove.path\" ).where(\"add is not null\" ).show(20, False)\\n--  Results\\n+-------------------------------------------------------------------+\\n|path                                                               |\\n+-------------------------------------------------------------------+\\n|part-00000-f661f932-f54f-4b38-8ed3-51f7770d3e55-c000.snappy.parquet|\\n+-------------------------------------------------------------------+\\nFigure 2-6. Delete data involves add file per V3 and V4\\nUnder the hood of a Delta Table | 67Notice how the remove.path  value is the same file as the preceding version 3 file. But\\nas you can see from the following query, a new file is created.\\n%python\\n# Add Information\\nj4.select(\"add.path\" ).where(\"add is not null\" ).show(20, False)\\n%scala\\n// Add Information\\nj4.select(\"add.path\" ).where(\"add is not null\" ).show(20, False)\\n-- Result\\n+-------------------------------------------------------------------+\\n|path                                                               |\\n+-------------------------------------------------------------------+\\n|part-00000-68bf51ea-6e00-4591-b267-62fcd36602ed-c000.snappy.parquet|\\n+-------------------------------------------------------------------+\\nIf you query this file, observe that it contains only the expected values of {3, 5-8} .\\n%python\\n# Read the V4 added file\\nspark.read.parquet(\"/table_delta/part-00000-68bf51ea-6e00-4591-\\nb267-62fcd36602ed-c000.snappy.parquet\" ).show()\\n%python\\n// Read the V4 added file\\nspark.read.parquet(\"/table_delta/part-00000-68bf51ea-6e00-4591-\\nb267-62fcd36602ed-c000.snappy.parquet\" ).show()\\n-- Result\\n+---+\\n| id|\\n+---+\\n|  8|\\n|  3|\\n|  6|\\n|  9|\\n|  7|\\n+---+\\nTo reiterate, instead of deleting the value 4 from the original file, instead Delta is cre‐\\nating a new file with the values not deleted (in this case) and keeping the old file in\\nplace.\\nLet’s go back in time.    What you’re seeing here is multiversion concurrency control\\n(MVCC) in action where there are two separate files associated with two different\\ntable versions. But how about if you do want to query the previous data? Well, this is\\nan excellent segue to Delta time travel options.\\n68 | Chapter 2: Time Travel with Delta LakeTime Travel\\nIn Chapter 1 we unpacked what was happening with the transaction log to provide\\nACID transactional semantics. In the previous section, we unpacked what was hap‐\\npening with the data files in relation to the transaction log in light of Delta transac‐\\ntional protocol with particular focus on multiversion concurrency control (MVCC).\\nAn interesting byproduct of MVCC is that all of the files from previous transactions\\nstill reside and are accessible in storage. That is, the positive byproduct of the imple‐\\nmentation of MVCC in Delta is data versioning  or time travel .\\nFrom a high-level, Delta automatically versions the big data that you store in your\\ndata lake, and you can access any historical version of that data. This temporal data\\nmanagement simplifies your data pipeline by making it easy to audit, roll back data in\\ncase of accidental bad writes or deletes, and reproduce experiments and reports. Y ou\\ncan standardize on a clean, centralized, versioned big data repository in your own\\ncloud storage for your analytics.\\nCommon Challenges with Changing Data\\nData versioning is an important tool to ensure data reliability to address common\\nchallenges with changing data.\\nAudit data changes\\nAuditing data changes is critical from both in terms of data compliance as well as\\nsimple debugging to understand how data has changed over time. Organizations\\nmoving from traditional data systems to big data technologies and the cloud\\nstruggle in such scenarios.\\nReproduce experiments & reports\\nDuring model training, data scientists run various experiments with different\\nparameters on a given set of data. When scientists revisit their experiments after\\na period of time to reproduce the models, typically the source data has been\\nmodified by upstream pipelines. Lot of times, they are caught unaware by such\\nupstream data changes and hence struggle to reproduce their experiments. Some\\nscientists and organizations engineer best practices by creating multiple copies of\\nthe data, leading to increased storage costs. The same is true for analysts generat‐\\ning reports.\\nRollbacks\\nData pipelines can sometimes write bad data for downstream consumers. This\\ncan happen because of issues ranging from infrastructure instabilities to messy\\ndata to bugs in the pipeline. For pipelines that do simple appends to directories\\nor a table, rollbacks can easily be addressed by date-based partitioning. With\\nupdates and deletes, this can become very complicated, and data engineers typi‐\\ncally have to engineer a complex pipeline to deal with such scenarios.\\nTime Travel | 69Working with Time Travel\\nDelta’s time travel capabilities simplify building data pipelines for the preceding use\\ncases which we will further explore in the following sections. As noted previously, as\\nyou write into a Delta table or directory, every operation is automatically versioned.\\nY ou can access the different versions of the data two different ways:\\n1.Using a timestamp\\n2.Using a version number\\nIn the following examples, we will use the generated TPC-DS dataset so we can work\\nwith the following example customer table ( customer_t1 ). To review the historical\\ndata, run the DESCRIBE HISTORY  command.\\n%sql\\nDESCRIBE  HISTORY customer_t1 ;\\nThe following results are an abridged version of the table initially focusing only on\\nthe version , timestamp , and operation  columns.\\n-- Abridged Results\\n+-------+-------------------+---------+\\n|version|          timestamp|operation|\\n+-------+-------------------+---------+\\n|     19|2020-10-30 18:15:03|    WRITE|\\n|     18|2020-10-30 18:10:47|   DELETE|\\n|     17|2020-10-30 08:41:47|    WRITE|\\n|     16|2020-10-30 08:37:29|   DELETE|\\n|     15|2020-10-30 05:34:29|  RESTORE|\\n|     14|2020-10-30 02:50:07|    WRITE|\\n|     13|2020-10-30 02:37:17|   DELETE|\\n|     12|2020-10-30 02:31:36|    WRITE|\\n|     11|2020-10-28 23:33:34|   DELETE|\\n+-------+-------------------+---------+\\nThe following sections describe the various techniques to work with your versioned\\ndata.\\nUsing a timestamp\\nY ou can provide the timestamp or date string as an option to the DataFrame reader\\nusing the following syntax.\\n%sql\\n-- Query metastore-defined Delta table by timestamp\\nSELECT * FROM my_table  TIMESTAMP  AS OF \"2019-01-01\"\\nSELECT * FROM my_table  TIMESTAMP  AS OF date_sub (current_date (), 1)\\nSELECT * FROM my_table  TIMESTAMP  AS OF \"2019-01-01 01:30:00.000\"\\n-- Query Delta table by file path by timestamp\\n70 | Chapter 2: Time Travel with Delta LakeSELECT * FROM delta.`<path-to-delta>` TIMESTAMP  AS OF \"2019-01-01\"\\nSELECT * FROM delta.`<path-to-delta>` TIMESTAMP  AS OF date_sub (current_date (), \\n1)\\nSELECT * FROM delta.`<path-to-delta>` TIMESTAMP  AS OF \"2019-01-01 01:30:00.000\"\\n%python\\n# Load into Spark DataFrame from Delta table by timestamp\\n(df = spark.read\\n  .format(\"delta\")\\n  .option(\"timestampAsOf\" , \"2019-01-01\" )\\n  .load(\"/path/to/my/table\" ))\\n%scala\\n// Load into Spark DataFrame from Delta table by timestamp\\nval df = spark.read\\n  .format(\"delta\")\\n  .option(\"timestampAsOf\" , \"2019-01-01\" )\\n  .load(\"/path/to/my/table\" )\\nSpecific to the customer_t1  table, the following queries will allow you to query by\\ntwo different timestamps. The following SQL queries will use the metastore-defined\\nDelta table syntax.\\n%sql\\n-- Row count as of timestamp t1\\nSELECT COUNT(1) FROM customer_t1  TIMESTAMP  AS OF \"2020-10-30T18:15:03.000+0000\"\\n-- Row count as of timestamp t2\\nSELECT COUNT(1) FROM customer_t1  TIMESTAMP  AS OF \"2020-10-30T18:10:47.000+0000\"\\n%python\\n# timestamps\\nt1 = \"2020-10-30T18:15:03.000+0000\"\\nt2 = \"2020-10-30T18:10:47.000+0000\"\\nDELTA_PATH =\"/demo/customer_t1\"\\n># Row count as of timestamp t1\\n(spark.read.format(\"delta\")\\n     .option(\"timestampAsOf\" , t1).load(DELTA_PATH ).count())\\n# Row count as of timestamp t2\\n(spark.read.format(\"delta\")\\\\\\n     .option(\"timestampAsOf\" , t2).load(DELTA_PATH ).count())\\n%scala\\n// timestamps\\nval t1 = \"2020-10-30T18:15:03.000+0000\"\\nval t2 = \"2020-10-30T18:10:47.000+0000\"\\nval DELTA_PATH =\"/demo/customer_t1\"\\n// Row count as of timestamp t1\\nspark.read.format(\"delta\")\\n     .option(\"timestampAsOf\" , t1).load(DELTA_PATH ).count()\\nTime Travel | 71// Row count as of timestamp t2\\nspark.read.format(\"delta\")\\n    .option(\"timestampAsOf\" , t2).load(DELTA_PATH ).count()\\nThe preceding queries have the following results below.\\nTimestamp Row Count\\n2020-10-30T18:15:03.000+0000 65000000\\n2020-10-30T18:10:47.000+0000 58500000\\nIf the reader code is in a library that you don’t have access to, and if you are passing\\ninput parameters to the library to read data, you can still travel back in time for a\\ntable by passing the timestamp in yyyyMMddHHmmssSSS  format to the path:\\nspark.re\\nad.format(\"delta\").load(<path-to-delta>@yyyyMMddHHmmssSSS)\\n            \\nFor example, to query for the timestamp 2020-10-30T18:15:03.000+0000 , use the\\nfollowing Python and Scala syntax.\\n%python\\n# Delta table base path\\nBASE_PATH =\"/demo/customer_t1\"\\n# Include timestamp in yyyyMMddHHmmssSSS format\\nDELTA_PATH =BASE_PATH  + \"@20201030181503000\"\\n# Get row count of the Delta table by timestamp using @ parameter\\nspark.read.format(\"delta\").load(DELTA_PATH ).count()\\n%scala\\n// Delta table base path\\nval BASE_PATH =\"/demo/customer_t1\"\\n// Include timestamp in yyyyMMddHHmmssSSS format\\nval DELTA_PATH =BASE_PATH  + \"@20201030181503000\"\\n// Get row count of the Delta table by timestamp using @ parameter\\nspark.read.format(\"delta\").load(DELTA_PATH ).count()\\nUsing a version number\\nIn Delta, every write has a version number, and you can use the version number to\\ntravel back in time as well.\\n%sql\\n-- Query metastore-defined Delta table by version\\nSELECT COUNT(*) FROM my_table  VERSION AS OF 5238\\nSELECT COUNT(*) FROM my_table @v5238\\n72 | Chapter 2: Time Travel with Delta Lake-- Query Delta table by file path by version\\nSELECT count(*) FROM delta.`/path/to/my/table@v5238`\\n%python\\n# Query Delta table by version using versionAsOf\\n(df = spark.read\\n  .format(\"delta\")\\n  .option(\"versionAsOf\" , \"5238\")\\n  .load(\"/path/to/my/table\" ))\\n# Query Delta table by version using @ parameter\\n(df = spark.read\\n  .format(\"delta\")\\n  .load(\"/path/to/my/table@v5238\" ))\\n%scala\\n// Query Delta table by version using versionAsOf\\nval df = spark.read\\n  .format(\"delta\")\\n  .option(\"versionAsOf\" , \"5238\")\\n  .load(\"/path/to/my/table\" )\\n// Query Delta table by version using @ parameter\\nval df = spark.read\\n  .format(\"delta\")\\n  .load(\"/path/to/my/table@v5238\" )\\nSpecific to the customer_t1  table, the following queries will allow you to query by\\ntwo different versions. The following SQL queries will use the metastore-defined\\nDelta table syntax.\\n%sql\\n-- Query metastore-defined Delta table by version 19\\nSELECT COUNT(1) FROM customer_t1  VERSION AS OF 19\\nSELECT COUNT(1) FROM customer_t1 @v19\\n-- Query metastore-defined Delta table by version 18\\nSELECT COUNT(1) FROM customer_t1  VERSION AS OF 18\\nSELECT COUNT(1) FROM customer_t1 @v18\\n%python\\n# Delta table base path\\nDELTA_PATH =\"/demo/customer_t1\"\\n# Row count of Delta table by version 19\\n(spark.read.format(\"delta\")\\n     .option(\"versionAsOf\" , 19).load(DELTA_PATH ).count())\\n# Row count of Delta table by version 18\\n(spark.read.format(\"delta\")\\n  .option(\"versionAsOf\" , 18).load(DELTA_PATH ).count())\\n# Row count of Delta table by @ parameter for version 18\\n(spark.read\\nTime Travel | 73  .format(\"delta\")\\n  .load(DELTA_PATH  + \"@v18\").count())\\n%scala\\n# Delta table base path\\nval DELTA_PATH =\"/demo/customer_t1\"\\n# Row count of Delta table by version 19\\nspark.read.format(\"delta\")\\n     .option(\"versionAsOf\" , 19).load(DELTA_PATH ).count()\\n# Row count of Delta table by version 18\\nspark.read.format(\"delta\")\\n     .option(\"versionAsOf\" , 18).load(DELTA_PATH ).count()\\n# Row count of Delta table by @ parameter  for version 18\\nspark.read\\n  .format(\"delta\")\\n  .load(DELTA_PATH  + \"@v18\").count()\\nThe preceding queries have the following results below.\\nVersion Row Count\\n19 65000000\\n18 58500000\\nTime travel use cases\\nIn this section, we will focus on how to apply time travel for various use cases, and\\nwhy it’s important to them.\\nDebug\\nTo troubleshoot the ETL pipeline or data quality issues, or to fix the accidental\\nbroken data pipelines. This is the most common use case for time travel and we\\nwill be exploring this use case throughout the book.\\nGovernance and Auditing\\nTime travel, offers a verifiable data lineage that is useful for governance, audit\\nand compliance purposes. As the definitive record of every change ever made to\\na table, you can trace the origin of an inadvertent change or a bug in a pipeline\\nback to the exact action that caused it. If your GDPR pipeline job had a bug that\\naccidentally deleted user information, you can fix the pipeline. Retention and\\nvacuum allows you to act on Data subject requests well within the timeframe.\\nRollbacks\\nTime travel also makes it easy to do rollbacks in case of bad writes. Due to a pre‐\\nvious error, it may be necessary to rollback to a previous version of the table\\nbefore continuing forward with data processing.\\n74 | Chapter 2: Time Travel with Delta LakeReproduce experiments & reports\\nTime travel also plays an important role in machine learning and data science.\\nReproducibility of models and experiments is a key consideration for data scien‐\\ntists, because they often create 100s of models before they put one into produc‐\\ntion, and in that time-consuming process would like to go back to earlier models.\\nPinned view of a continuously updating Delta table across multiple downstream jobs\\nWith AS OF queries, you can now pin the snapshot of a continuously updating\\nDelta table for multiple downstream jobs.\\nTime series analytics\\nTime travel simplifies time series analytics. Y ou can use timestamp  and As Of\\nqueries to extract meaningful statistics associated with the selected data points\\nand shift in associated variables over time.\\nNext we will provide further details for each use case. Note, we will cover more\\nadvanced use cases which include time travel in chapter 14.\\nUse Case: Governance and Auditing\\nThe use case of Governance described here is about protecting people’s information\\nin regards to GDPR (General Data Protection Regulation) and CCPA (California\\nConsumer Privacy Act). The role of governance is to set the direction of protecting\\ninformation in the form of policies, standards, and guidelines.\\nOne of the most common scenarios surrounding governance to protect people’s\\ninformation is the need to perform a delete request as part of any GPDR or CCPA\\ncompliance. What that means for data engineers is that you need to locate the records\\nfrom your Delta Lake and delete it within the specified time period. Users can run\\nDESCRIBE HISTORY to see metadata around the changes that were made. An example\\nof this scenario is how Starbucks  implements governance, risk and control for their\\nDelta lake.\\nLet’s dive further in the simplified  governance use case (see Figure 3-x)..   \\nTime Travel | 75Figure 2-7. Simplified  governance use case with time travel\\n1.A delete request is recorded in the system where the user has requested that their\\ninformation be removed from the data lake. This information is often recorded\\nin its own table or store for auditing and lineage purposes.\\n2.This table ( customer_delete_keys ) will need to join with the customer table\\n(customer_t1 ) to identify which users in the table need to be removed.\\n3.Y ou will then delete this data from the system to address this delete request.\\nWith Delta Lake, you can achieve the above listed goals by running the following\\nsteps. Y ou can also use this notebook  for the whole code.\\nIdentify users that need to be removed\\nAs noted earlier, the customer_delete_keys table contains the set of customer keys\\nthat need to be removed. To review the table, run the following query.\\n%sql\\nSELECT * FROM customer_delete_keys\\n%python\\n# Query table using metastore defined table\\nsql(\"SELECT * FROM customer_delete_keys\" ).show()\\n# Query table from file path\\nspark.read.format(“delta”).load(\"/customer_delta_keys\" ).show()\\n%scala\\n// Query table using metastore defined table\\nsql(\"SELECT * FROM customer_delete_keys”).show()\\n76 | Chapter 2: Time Travel with Delta Lake// Query table from file path\\nspark.read.format(“delta”).load(\" /customer_delta_keys \").show()\\nNote, going forward in this section, we will only use metastore-\\ndefined SQL queries. In Python and Scala, you can also use the syn‐\\ntax:\\nsql(\"<sql-statement-here>\").show()\\nA partial result of the above queries can be seen in the output below.\\n+----------------+-------------+\\n|   c_customer_id|c_customer_sk|\\n+----------------+-------------+\\n|AAAAAAAAKHJBCKBA|     27400570|\\n|AAAAAAAAFIJBCKBA|     27400581|\\n|AAAAAAAAGIJBCKBA|     27400582|\\n|AAAAAAAAHIJBCKBA|     27400583|\\n|...             |          ...|\\n+----------------+-------------+\\nThe following SQL query will allow you to quickly identify the records that need to\\nbe removed from customer_t1 .\\n%sql\\n-- Identify the records\\nSELECT COUNT(*)\\n  FROM customer_t1\\n WHERE c_customer_id  IN (\\n  SELECT c_customer_id  \\n    FROM customer_delete_keys\\n)\\n-- Result\\n+--------+\\n|count(1)|\\n+--------+\\n| 6500000|\\n+--------+\\nDelete the identified  users\\nOnce we have identified the matches, you can now go ahead and use the DELETE\\nfunction to remove those records from the Delta table.\\n%sql\\n-- Delete identified records\\nDELETE\\n  FROM customer_t1\\n WHERE c_customer_id  IN (\\n  SELECT c_customer_id  \\nTime Travel | 77    FROM customer_delete_keys\\n)\\nFortunately this process is pretty straightforward as you can remove the SELECT\\nclause and replace it with a DELETE  clause. Y ou can validate this by reviewing the\\ntransaction log via the DESCRIBE HISTORY  command (See Figure 3-x).\\n%sql\\nDESCRIBE  HISTORY customer_t1 ;\\nFigure 2-8. Reviewing history after  fulfilling  GDPR delete request\\nBelow is the transposed table of the DELETE transaction so we can inspect informa‐\\ntion.\\nColumn Name Column Value\\nversion 13\\noperation DELETE\\noperationParameters [predicate -> [\"exists(t1.`c_customer_id`)\"]]\\noperationMetrics [\\nnumRemovedFiles -> 23, \\nnumDeletedRows -> 6500000, \\nnumAddedFiles -> 100, \\nnumCopiedRows -> 42250002\\n]\\nLet’s dive further in the audit use case.    Auditor’s role is to review and inspect that deliv‐\\nery conforms to those guidelines. For auditing purposes, you can review the Delta\\ntransaction log which tracks all of the changes to your Delta table. In this case, for\\nversion 13 of the table:\\n•Through the operation  column, a DELETE  activity was recorded.\\n•The operationParameters  denote how the delete occurred, i.e. the delete was\\ndone by the c_customer_id  column.\\n•From an auditing perspective, the important callout is the 6,500,000 rows deleted\\n(numDeletedRows ) under operationMetrics . This allows us to determine the\\nmagnitude of impact on the records.\\n78 | Chapter 2: Time Travel with Delta LakeDelete is a tombstone.    It is important to note that when a DELETE  is executed on a\\nDelta table, the most current version of the table no longer reads this data, as it is\\nexplicitly excluded. If you recall earlier in the chapter, the Remove  column of the\\ntransaction log will contain the files that no longer should be included.\\nFor some organizations, this meets their compliance requirements as the data is no\\nlonger accessible by default. But depending on your legal requirements, there are\\nadditional considerations:\\n•If the data must be removed immediately, run the VACUUM  command to remove\\nthe data files\\n•Instead of deleting, UPDATE  the data (e.g. update the name column with some\\nother value such as a GUID) so you can keep the associated IDs without having\\nany personally identifiable information.\\n•Place personally identifiable information into a separate table so that GDPR\\ndelete requests would result in only updating or deleting the information from\\nthis demographics table with little to no impact on your fact data.\\nIt is important to note that this example is showcasing a simplified governance use\\ncase. Y ou will need to work with your legal department to determine your actual\\nrequirements. The key takeaway for our discussion here is that Delta Lake time travel\\nallows you to safely remove data from your Delta Lake, reverse it in case there are\\nmistakes, and track the progress through the Delta Lake transaction log.\\nFor more information on governance:\\n•Tech Talk | Addressing GDPR and CCPA Scenarios with Delta Lake and Apache\\nSpark\\n•Data+AI Summit 2020 | Operationalizing Big Data Pipelines At Scale at Star‐\\nbucks\\n•Webinar | Is Y our Data Lake GDPR Ready? How to Avoid Drowning in Data\\nRequests\\nUse Case: Rollbacks\\nTime travel also makes it easy to do rollbacks so you can reset your data to a previous\\nsnapshot. If a user accidentally inserted new data or your pipeline job inadvertently\\ndeleted user information, you can easily fix it by restoring the previous version of the\\ntable.\\nThe following code sample takes our existing customer_t1  table and restores it to\\nversion 17. Let’s start by reviewing the history of our table.\\nTime Travel | 79%sql\\nDESCRIBE  HISTORY customer_t1\\n--- abridged results\\n+-------+---------+---------------------------------------------+\\n|version|operation|operationParameters                          |\\n+-------+---------+---------------------------------------------+\\n|18     |DELETE   |[predicate -> [\"exists(t1.`c_customer_id`)\"]]|\\n|17     |WRITE    |[mode -> Overwrite, partitionBy -> []]       |\\n+-------+---------+---------------------------------------------+\\nInsert and overwrite.    To rollback, what we really are doing is that once we identify\\nwhich version of the table we want to travel back to, we will load that version using as\\nof. Behind the scenes, delta inserts it as a new version and overwrites the latest\\nsnapshot. In this scenario, you can see from the abridged history results that version\\n18 of the table contains a DELETE  operation that we want to rollback from. The follow‐\\ning code allows you to perform the rollback.\\n%python\\n# Restore customer_t1 table to Version 17\\n># Load Version 17 data using df\\ndf = spark.read.format(\"delta\").option(\"versionAsOf\" , \"17\").load(\"/customer_t1\" )\\n# Overwrite Version 17 as the current version\\ndf.write.format(\"delta\").mode(\"overwrite\" ).save(\"/customer_t1\" )\\n%scala\\n// Restore customer_t1 table to Version 17\\n// Load Version 17 data using df\\nval df = spark.read.format(\"delta\").option(\"versionAsOf\" , \"17\").load(\"/\\ncustomer_t1\" )\\n// Overwrite Version 17 as the current version\\ndf.write.format(\"delta\").mode(\"overwrite\" ).save(\"/customer_t1\" )\\nY ou can validate this by either querying the data or reviewing the table history. The\\nfollowing is how you would do this using SQL.\\n%sql\\n-- Get the row counts between two table versions and calculate the difference\\n SELECT a.v19_cnt, b.v17_cnt, a.v19_cnt - b.v17_cnt AS difference\\n   FROM\\n -- Get V19 row count\\n(SELECT COUNT(1) AS v19_cnt\\n   FROM customer_t1  VERSION AS OF 19) a\\n  CROSS JOIN\\n -- Get V17 row count\\n(SELECT COUNT(1) AS v17_cnt\\n>   FROM customer_t1  VERSION AS OF 17) b\\n-- Result\\n+--------+--------+----------+\\n| v19_cnt| v17_cnt|difference|\\n80 | Chapter 2: Time Travel with Delta Lake+--------+--------+----------+\\n|65000000|65000000|         0|\\n+--------+--------+----------+\\nA comparable code can be written in Python and Scala versions as shown below .\\n%python\\n# Get V19 row count\\nv19_cnt = sql(\"SELECT COUNT(1) AS v19_cnt FROM customer_t1 VERSION AS OF \\n19\").first()[0]\\n# Get V17 row count\\nv17_cnt = sql(\"SELECT COUNT(1) AS v17_cnt FROM customer_t1 VERSION AS OF \\n17\").first()[0]\\n# Calculate the difference\\nprint(\"Difference in row count between v19 and v17: %s\" % (v19_cnt - v17_cnt))\\n%scala\\n// Get V19 row count\\nval v19_cnt = sql(\"SELECT COUNT(1) AS v19_cnt FROM customer_t1 VERSION AS OF \\n19\").first()(0).toString .toInt\\n// Get V19 row count\\nval v17_cnt = sql(\"SELECT COUNT(1) AS v17_cnt FROM customer_t1 VERSION AS OF \\n17\").first()(0).toString .toInt\\n// Calculate the difference\\nprint(\"Difference in row count between v19 and v17: \"  + (v19_cnt - v17_cnt))\\nY ou will get the same result for both the Python and Scala examples.\\n-- Result\\nDifference in row count between v19 and v17: 0\\nY ou can also validate the restore by reviewing the table history; note that there is now\\nversion 19 that records the overwrite statement.\\n%sql\\nDESCRIBE HISTORY customer_t1\\n--- abridged results\\n +-------+---------+---------------------------------------------+\\n|version|operation|operationParameters                          |\\n+-------+---------+---------------------------------------------+\\n|19     |WRITE    |[mode -> Overwrite, partitionBy -> []]       |\\n|18     |DELETE   |[predicate -> [\"exists(t1.`c_customer_id`)\"]]|\\n|17     |WRITE    |[mode -> Overwrite, partitionBy -> []]       |\\n+-------+---------+---------------------------------------------+\\nRESTORE command.    Note, if you are using Databricks, this step will be a little easier\\nwhen restoring with at least DBR 7.4 with the RESTORE command . Y ou can perform\\nthe same steps as above with a single atomic command.\\nTime Travel | 81%sql\\n-- Restore table as of version 17\\nRESTORE customer_t1  VERSION AS OF 17\\nIn addition to the previous queries comparing the count of two different versions,\\nonce you run the RESTORE , additional information gets stored within the transaction\\nlog which are shown under the operationMetrics  column map.\\n%sql\\nDESCRIBE  HISTORY customer_t1\\n-- Abridged results\\n+-------+---------+---------------------------------------------+\\n|version|operation|operationParameters                          |\\n+-------+---------+---------------------------------------------+\\n|20     |RESTORE  |[version -> 17, timestamp ->]                |\\n|19     |WRITE    |[mode -> Overwrite, partitionBy -> []]       |\\n|18     |DELETE   |[predicate -> [\"exists(t1.`c_customer_id`)\"]]|\\n|17     |WRITE    |[mode -> Overwrite, partitionBy -> []]       |\\n+-------+---------+---------------------------------------------+\\nWe are highlighting the record 20 from our output above for the\\nrestore  operation of the delta table to explain the operationMet\\nrics  restore-based parameters.\\n{\\n\"numRestoredFiles\": \"31\", \\n\"removedFilesSize\": \"3453647999\", \\n\"numRemovedFiles\": \"31\", \\n\"restoredFilesSize\": \"3453647999\",\\n\"numOfFilesAfterRestore\": \"31\", \\n\"tableSizeAfterRestore\": \"3453647999\"\\n}\\nFor more information about the RESTORE command, see Restore a Delta table .\\nTime travel considerations\\nWe need to keep additional considerations in mind with respect to time travel.\\nData retention:\\n•One very important thing to remember is Data retention. Y ou can only travel\\nback so far depending on what time limits we set and how much data is retained.\\n•To time travel to a previous version, you must retain both the log and the data\\nfiles for that version.\\n82 | Chapter 2: Time Travel with Delta LakeVacuum :\\n•The ability to time travel back to a version older than the retention period is lost\\nafter running vacuum.\\n•By default, `vacuum()` retains all the data needed for the last 7 days.\\n•When you run the vacuum  with specified retention of 0 hours, it will throw an\\nexception.\\n—Delta protects you by giving this warning to save you from permanent\\nchanges to your data or if a downstream reader or application is referencing\\nyour table.\\n—If you are certain that there are no operations being performed on this table,\\nsuch as insert/upsert/delete/optimize, then you may turn off this check by set‐\\nting: \\nspark.databricks.delta.retentionDurationCheck.enabled = false\\n•VACUUM  doesn’t clean up log files; log files are automatically cleaned up after\\ncheckpoints are written.\\n•Another note to satisfy the delete requirement is that you also need to delete it\\nfrom your blob storage. So as a best practice, whenever the ETL pipeline is set\\nyou should apply the retention policy from the start itself so you don’t have to\\nworry later.\\nSummary\\nIn this chapter, we focussed on a powerful benefit of the MVCC (Multi-Version Con‐\\ncurrency Control) implementation of Delta, that enables users to have the benefit of\\ndata versioning or time travel . In addition to showing common challenges with\\nchanging data, we explained the Delta time machine and its properties, how time\\ntravel works and what operations you can perform while traveling back in time.\\nAs alluded to earlier in the chapter, time travel allows developers to debug what hap‐\\npened with their Delta table, improving developer productivity tremendously. One of\\nthe common reasons developers need to debug their table is due to data modification\\nsuch as delete, update, and merge. This brings us to our next chapter, Chapter 4: Data\\nModifications in Delta tables.\\nSummary | 83CHAPTER 3\\nContinuous Applications with Delta Lake\\nA Note for Early Release Readers\\nWith Early Release ebooks, you get books in their earliest form—the authors’ raw and\\nunedited content as they write—so you can take advantage of these technologies long\\nbefore the official release of these titles.\\nThis will be the 6th chapter of the final book. Please note that the GitHub repo will be\\nmade active later on.\\nIf you have comments about how we might improve the content and/or examples in\\nthis book, or if you notice missing material within this chapter, please reach out to the\\neditor at gobrien@oreilly.com .\\nIn the previous chapters, we discussed the Delta transaction log (Chapter 2), time\\ntravel (Chapter 3), schema enforcement and evolution (Chapter 4) and data modifi‐\\ncations (Chapter 5). While this was never explicitly stated, one of the primary reasons\\nfor creating Delta Lake was to provide atomic transactionality  for continuous applica‐\\ntions. At its most fundamental level, the concept of continuous applications is the\\nability to develop streaming applications without the need to reason  for streaming\\napplications. In this chapter, we will initially provide the background of streaming\\napplications, the importance of placing structure around streaming, and how this\\nprogression of structured streaming eventually led to the development of Delta Lake.\\nIn the process of creating structured streaming, for developers to write application\\nlogic not defined by if the code was streaming or batch processing, we would need to\\ndecouple business logic from latency. It is then ironic that we needed to pull a concept\\nfrom tightly coupled relational database systems to perform this task - to bring back\\nACID transactions - hence the development of Delta Lake.\\n85Note, this chapter will review high-level concepts of Spark Streaming and Spark\\nStructured Streaming including example code. But, it will not provide an in-depth\\nreview of all of these concepts; for more information, please refer to Learning Spark\\n2nd Edition,  Chapter 8: Structured Streaming and Continuous Applications. Instead,\\nthis chapter will provide a retrospective from streaming to continuous applications\\nand how these concepts helped define what is now Delta Lake.\\nMake All Your Streams Come True\\nLet’s go back in time to the early 2010s where the context of streaming and complex\\nevent processing was with the Apache Storm project. Note, other open-source and\\nproprietary streaming projects did exist at this time but due to the popularity of Twit‐\\nter and the open-sourcing of the Storm project, Apache Storm stormed  onto the\\nstreaming scene (sorry, not sorry for the pun). So at least from an open-source big\\ndata perspective, the Storm project was the baseline for development of streaming.\\nThe benefits of streaming were that it was tuned for lower latency with reasonable\\nthroughput for lower or medium size data. But as this was the nascent age of Big\\nData, what was desired was the ability to have higher throughput and lower latency\\nfor larger amounts of data. Because a streaming system like Storm was continuously\\nprocessing data, you could achieve performing an event processing against large\\namounts of data since the event processing itself was against a small amount of data.\\nA simple example of this would be an onclick  event where when a user clicked on a\\nlink, this would trigger some other process downstream (e.g. As an ad-business, you\\nprovide an advertisement specific to the link(s) the user clicked).\\nAchieving event processing with state proved to be a much more complicated task.\\nCalculating aggregates against continuous streams of data required memory and file\\nmanagement, moreover doing so in a reliable manner became exceedingly difficult. A\\nsimple example of this would be for you to calculate across hourly time windows how\\nmany users had clicked on a link. Because the aggregations could become exceedingly\\ncomplex and expensive, typically this was achieved by building separate pipelines to\\nprocess the event processing data while another pipeline to process the state or aggre‐\\ngate data. This latter design led to the creation of the lambda architecture  which\\nallowed us to build two different systems as defined by the batch/serving and speed\\nlayers. This enabled developers to choose and customize their frameworks to opti‐\\nmize for each respective layer. For more details on the lambda architecture, refer to\\nChapter 7: Medallion Architecture.\\nThe problem with this approach was that it required building and maintaining two\\ndifferent pipelines. These pipelines were often created by two different teams with\\ndifferent programming paradigms; for example, the speed layer was addressed by\\nusing storm-clojure  while the batch/server layer was addressed by using Hadoop and\\n86 | Chapter 3: Continuous Applications with Delta LakeHive jobs. These differences prevented utilizing the same team to address these prob‐\\nlems and often required a third team to reconcile these two layers.\\nTo simplify this, Spark Streaming was built to unify these two layers. Let’s deep dive\\ninto this in the next section.\\nSpark Streaming Was Built to Unify Batch and Streaming\\nMany developers in the Big Data community were already familiar with and using\\nApache Spark™ for batch processing large amounts of data. Spark Streaming is an\\nextension of the core Spark API that enables scalable, high-throughput, fault-tolerant\\nstream processing of live data streams.\\nThe benefits of using the same framework to address both speed and batch process‐\\ning were immense including:\\n•A high-level API  that simplified and extended event processing functionality\\nincluding joins and windows often with significantly less code.\\n•Spark streaming is fault-tolerant  with exactly-once semantics even for stateful\\nops (more on exactly-once semantics later in the chapter)\\n•Because it is Spark, there were the benefits of integration  with other features of\\nSpark: Machine Learning with MLlib, SQL and DataFrames, and graph process‐\\ning with GraphX to name a few.\\nFigure 3-1. Spark Streaming\\nAs noted previously, for more information on Spark Streaming, refer to Learning\\nSpark 2nd Edition , Chapter 8: Structured Streaming and Continuous Applications or\\nSpark Streaming - Spark 3.0.1 Documentation (apache.org) .\\nMake All Your Streams Come True | 871See Spark Streaming: What Is It and Who’s Using It?What are the primary use cases for streaming?\\nFor many, Spark Streaming opened the door for a new paradigm in data processing as\\nmany data practitioners had focused on batch processing. As our fellow author Tatha‐\\ngata Das noted1:\\nMost companies are collecting more data than ever and wanting to get value from\\nthat data in real-time. Whether it was sensors, IoT devices, social networks, or online\\ntransactions, they all generated a massive amount of data that needs to be monitored\\nconstantly and acted upon quickly. As a result, the need for large-scale, real-time\\nstream processing has become more evident than ever before.\\nFigure 3-2. Example streaming use cases\\nThat same streaming data is likely collected and used in batch jobs when generating\\ndaily reports and updating models. This means that a modern stream processing\\npipeline needs to be built, taking into account not just the real-time aspect, but also\\nthe associated pre-processing and post-processing aspects (e.g. model building).\\nBefore Spark Streaming, building complex pipelines that encompass streaming,\\nbatch, or even machine learning capabilities with open source software meant dealing\\nwith multiple frameworks, each built for a niche purpose, such as Storm for real-time\\nactions, Hadoop MapReduce for batch processing, etc. Besides the pain of developing\\ndisparate programming models, there was a huge cost of managing multiple frame‐\\n88 | Chapter 3: Continuous Applications with Delta Lakeworks in production. Spark and Spark Streaming, with its unified programming\\nmodel and processing engine, makes all of this very simple.\\nThus, if you were to view the data world from the lens of latency, then we could view\\nthis purely from the standpoint of traditional streaming systems (e.g. alerting) and\\ntraditional batch systems (e.g. ETL).\\nFigure 3-3. Streaming and batch systems started overlapping\\nBut these primary use cases started overlapping as traditional batch systems required\\nlower latencies and traditional streaming systems were required to deal with larger\\nvolumes (e.g. aggregations). This is how Spark Streaming unified batch and stream\\nprocessing because it allows you to solve the continuum of real-time analytics.\\nMake All Your Streams Come True | 89Use case highlights\\nAs they noted in their blog Can Spark Streaming Survive Chaos Monkey , Netflix\\nreceives billions of events per day from various sources, and they have used Kafka\\nand Spark Streaming to build a real-time engine that provides movie recommenda‐\\ntions to its users.\\nFigure 3-4. Netflix’s  use of Spark Streaming (Source:  Can Spark Streaming Survive\\nChaos Monkey )\\nAs noted in the Spark+AI Summit session Spark and Spark Streaming at Netflix , back\\nin 2015, Netflix processed over 450 billion events/day with a peak of 8 million events/\\nsecond (approx 17GB). A great quote from Netflix concerning Spark Streaming can\\nbe seen below.\\nOverall, we are happy with the resiliency of spark standalone for our use cases and\\nexcited to take it to the next level where we are working towards building a unified\\nLambda Architecture that involves a combination of batch and real-time streaming\\nprocessing.\\nAs the preceding quote calls out nicely, Spark Streaming allowed Netflix and other\\norganizations to build a unified architecture that combined traditional streaming and\\nbatch processing.\\n90 | Chapter 3: Continuous Applications with Delta LakeFigure 3-5. Spark Streaming started to unify streaming and batch processing\\nSpark Streaming lacked structure\\nWhile Spark Streaming started unifying the concepts of streaming and batch process‐\\ning, it could not take advantage of the architectural changes of the Apache Spark™\\nproject. For starters, while streaming and batch were using the same framework, they\\nwere using different APIs - DStreams  and RDDs  respectively. At the same time,\\nDStreams  were not able to take advantage of the performance improvements of pro‐\\nviding structure to Apache Spark including but not limited to the new structured file\\nformats (e.g. Parquet, ORC, etc.), Spark SQL, Spark Catalyst optimizer, Project Tung‐\\nsten. For more information on this, please refer to:\\n•Learning Spark 2nd Edition\\n•Structuring Spark: DataFrames, Datasets, and Streaming by Michael Armbrust .\\nMake All Your Streams Come True | 91•Deep Dive into Spark SQL ’s Catalyst Optimizer\\n•Apache Spark as a Compiler: Joining a Billion Rows per Second on a Laptop .\\nSimply put, Spark Streaming itself was not designed for structured data. Before we\\ndiscuss how to provide structure to streaming - i.e. Structured Streaming - let’s also\\ndiscuss an important concept: exactly-once semantics.\\nExactly-Once Semantics\\nThe desired result of data processing - irrespective of latency - is exactly-once seman‐\\ntics. In other words, for every record end-to-end from source to target, there is nei‐\\nther duplicates nor any missing or incomplete data.\\nLet’s take an example to explain this in detail.\\nIn the figure above, there are multiple streams from the source to the Processing\\nEngine.\\n92 | Chapter 3: Continuous Applications with Delta Lake1.The first stream shows a complete successful event\\n2.The second one in red, shows a failure at the Processing engine level. If the fail‐\\nures occur, the end result is that the processing engine should account for each\\nrecord once.\\n3.When finally writing the data out for storage, this must be done in an idempotent\\nmanner so that there is only one copy of this record.\\nWhile this may be an obvious requirement in building and maintaining batch data\\npipelines, but for streaming pipelines it was not. Let’s summarize :\\n•Storm provided at-least once: no missing data, but duplicates could possibly\\noccur.\\n•The Spark Streaming solved (2) but the writes were not idempotent (3) from the\\nabove listed scenarios\\n•The Structured streaming solves both. It started getting some of the ACID\\nsemantics back for inserts / appends to a Parquet table\\nMake All Your Streams Come True | 93Putting Some Structure Around Streaming\\nIn Apache Spark 2.0, adding structure to Spark, through use of high-level Data‐\\nFrames and Datasets APIs, accommodates a novel approach to look at real-time\\nstreaming. That is, look at streaming not as streaming but as either a static table of\\ndata (where you know all the data) or a continuous table of data (where new data is\\ncontinuously arriving).\\nAs such you can build end-to-end continuous applications, in which you can issue\\nthe same queries to batch as to real-time data, perform ETL, generate reports, update\\nor track specific data in the stream. This combined batch & real-time query-\\ncapabilities to a structured stream is a unique offering—not many streaming engines\\noffer it yet. This can reduce latency and allow for incremental processing.\\nAt a conceptual level, Structured Streaming treats live data as a table that is being\\ncontinuously appended. The best thing about Structured Streaming is that it allows\\nyou to rapidly get value out of streaming systems with virtually no code changes.\\n94 | Chapter 3: Continuous Applications with Delta LakeStructured streaming allowed us to solve almost all the problems with ACID transac‐\\ntions. However building the pipelines with structured streaming only solves part of\\nthe problem. It solved the lack of ACID guarantees for inserts but data pipelines\\nrequire cleaning up data, GDPR deletes, data reprocessing tasks are still a challenge.\\nWhen reprocessing files in a directory, it is very easy to\\n•delete files that you need,\\n•Put them in the wrong folder, so the table cannot reconcile\\n•Partially written/corrupt files.\\nHere are more documents to get started with Structured streaming.\\n•Structured Streaming documentation\\n•Blog: Structured Streaming: A New High-Level API for\\nStreaming\\n• Docs: Structured Streaming: Introductory Notebooks\\n• Docs: Structured Streaming in Production\\nThis led to the idea of Delta Lake to ensure that all the changes are done in an atomic\\nmanner. More about this property was covered in detail in the Chapter 2, if you may\\nrecall.\\nMake All Your Streams Come True | 95A table in Delta Lake is a batch table as well as a streaming source and sink. Delta\\nLake uses Apache Parquet  files to store your data, and it is fully compatible with\\nApache Spark APIs .\\nStreaming with Delta\\nDelta Lake is tightly integrated with Spark Structured Streaming  through readStream\\nand writeStream . Some of the challenges that Delta solves in the context of stream‐\\ning are:\\n•Coalescing small files produced by low latency ingest\\n•Maintaining “exactly-once” processing with more than one stream (or concurrent\\nbatch jobs)\\n•Efficiently discovering which files are new when using files as the source for a\\nstream\\n96 | Chapter 3: Continuous Applications with Delta LakeAs described in one of our blogs with an advertising firm who serves data-driven\\nonline video ads for its clients, they leverage Structured Streaming, Delta Lake, and\\nDatabricks  for the brand safety. The diagram below shows how Eyeview implemented\\nthe concept of brand safety and what challenges the company faced in doing so.\\nKey highlights of their use case was that they were getting numerous URLs every sec‐\\nond and it can create following challenges while batch processing a large number of\\nsmall files.\\n•Challenge #1: Underutilized Resources\\n•Challenge #2: Manual Checkpointing\\n•Challenge #3: Parquet Table Issues\\n•Challenge #4: No Concurrent Reads on Parquet Tables\\nTo solve the above challenges, Eyeview introduced two new technologies: Spark\\nStreaming and Delta Lake. The source of most of a large number of small files can be\\nconverted from batch processing to streaming processing.\\n1.Spark streaming helped in solving the first two challenges. Instead of a cluster of\\nbidders writing files which contain the URLs to S3, they started sending URLs\\ndirectly to a kinesis stream.\\n2.By connecting Spark Streaming with Kinesis streams they no longer need to do\\nmanual checkpointing. Since Spark Streaming is inherently fault-tolerant they\\ndidn’t have to worry about failures and reprocessing of files.\\nThe code snippet below reads the data from the Kinesis stream.\\nimport org.apache.spark.sql.types._\\nval jsonSchema = new StructType()\\n .add(\"entry\", StringType)\\n .add(\"ts\", LongType)\\nval kinesisDF = spark.readStream\\n .format(\"kinesis\")\\n .option(\"streamName\", \"kinesis Stream Name\")\\n .option(\"initialPosition\", \"LATEST\")\\n .option(\"region\", \"aws-region\")\\n .load()\\nval queryDf = kinesisDF\\n .map(row => Charset.forName(\"UTF-8\").newDecoder().decode(ByteBuffer.wrap(new \\nBase64().decode(row.get(1)).asInstanceOf[Array[Byte]])).toString)\\n .selectExpr(\"cast (value as STRING) jsonData\")\\n .select(from_json(col(\"jsonData\"), jsonSchema).as(\"bc\"))\\n .withColumn(\"entry\",lit($\"bc.entry\"))\\n .withColumn(\"_tmp\", split($\"entry\", \"\\\\\\\\,\"))\\n .select(\\n $\"_tmp\".getItem(0).as(\"device_type\"),\\n $\"_tmp\".getItem(1).as(\"url\"),\\nMake All Your Streams Come True | 97 $\"_tmp\".getItem(2).as(\"os\"),\\n $\"bc.ts\".as(\"ts\")\\n).drop(\"_tmp\")\\n1.The other two challenges with the parquet table are solved by introducing a new\\ntable format, Delta Lake. Delta Lake supports ACID transactions, which basically\\nmeans they can concurrently and reliably read/write this table.\\n2.Delta Lake tables are also very efficient with continuous appends to the tables. A\\ntable in Delta Lake is both a batch table, as well as a streaming source and sink.\\nThe below code shows persisting the data into Delta Lake. This also helped us in\\nremoving the millisecond partitions. See the below code for reference. (Partitions are\\nup to only hour level.)\\nval sparkStreaming = queryDf.as[(String,String,String,Long)].mapPartitions{parti\\ntion=>\\n val http_client = new HttpClient\\n http_client.start\\n val partitionsResult = partition.map{record=>\\n try{\\n  val api_url = ApiCallingUtility.createAPIUrl(record._2,record._1,record._3)\\n  val result = ApiCallingUtility.apiCall(http_client.newRequest(api_url).time\\nout(500, TimeUnit.MILLISECONDS).send(),record._2,record._1)\\n  aerospikeWrite(api_url,result)\\n  result\\n }\\n catch{\\n  case e:Throwable=>{\\n  println(e)\\n  }\\n } \\n }\\n partitionsResult\\n}\\nOnce they moved the architecture from batch processing to a streaming solution,\\nthey were able to reduce the cluster size of the Spark jobs, thus significantly reducing\\nthe cost of the solution. More impressively, they only required one job to take care of\\nall the brand safety providers, which further reduced costs.\\n98 | Chapter 3: Continuous Applications with Delta LakeLets cover some more concepts of Delta streaming. We learned that a Delta table can\\nbe both - a stream source as well as sink.\\nDelta as a Stream Source\\nWhen you load a Delta table as a stream source and use it in a streaming query, the\\nquery processes all of the data present in the table as well as any new data that arrives\\nafter the stream is started.\\nY ou can load both paths and tables as a stream.\\n%Scala\\nspark.readStream .format(\"delta\")\\n  .load(\"/mnt/delta/events\" )\\nimport io.delta.implicits ._\\nspark.readStream .delta(\"/mnt/delta/events\" )\\nor\\n%Scala\\nimport io.delta.implicits ._\\nspark.readStream .format(\"delta\")\\n  .table(\"events\" )\\nSome helpful configuration tips:\\n•Y ou can control the maximum size of any micro-batch that Delta Lake gives to\\nstreaming by setting the maxFilesPerTrigger option. This specifies the maximum\\nnumber of new files to be considered in every trigger. The default is 1000.\\n•Rate-limit how much data gets processed in each micro-batch by setting the max‐\\nBytesPerTrigger option. This sets a “soft max” , meaning that a batch processes\\napproximately this amount of data and may process more than the limit. If you\\nuse Trigger.Once for your streaming, this option is ignored. If you use this option\\nin conjunction with maxFilesPerTrigger, the micro-batch processes data until\\neither the maxFilesPerTrigger or maxBytesPerTrigger limit is reached.\\nIn cases when the source table transactions are cleaned up due to\\nthe logRetentionDuration configuration and the stream lags in\\nprocessing, Delta Lake processes the data corresponding to the lat‐\\nest available transaction history of the source table but does not fail\\nthe stream. This can result in data being dropped.\\nIgnore Updates and Deletes\\nAs discussed in an earlier section. Structured streaming can handle ingest but does\\nnot handle input that is not an append and throws an exception if any modifications\\nMake All Your Streams Come True | 99occur on the table being used as a source. There are two main strategies for dealing\\nwith changes that cannot be automatically propagated downstream:\\n•Y ou can delete the output and checkpoint and restart the stream from the begin‐\\nning.\\n•Y ou can set either of these two options:\\n—ignoreDeletes: ignore transactions that delete data at partition boundaries.\\n—ignoreChanges: re-process updates if files had to be rewritten in the source\\ntable due to a data changing operation such as UPDATE, MERGE INTO,\\nDELETE (within partitions), or OVERWRITE. Unchanged rows may still be\\nemitted, therefore your downstream consumers should be able to handle\\nduplicates. Deletes are not propagated downstream. ignoreChanges subsumes\\nignoreDeletes. Therefore if you use ignoreChanges, your stream will not be\\ndisrupted by either deletions or updates to the source table.\\nTo explain this through an example, lets assume that you have user_events table\\nwith date, user_email, and action as the columns that is partitioned by date . As a part\\nof GDPR you need to delete data from the user_events table.\\nWhen you delete at partition boundaries, the files are already segmented by value so\\nthe delete just drops those files from the metadata. Thus, if you just want to delete\\ndata from some partitions, you can use:\\n%scala\\nspark.readStream .format(\"delta\")\\n  .option(\"ignoreDeletes\" , \"true\")\\n  .load(\"/mnt/delta/user_events\" )\\nHowever, if you have to delete data based on user_email, then you will need to use:\\n%scala\\nspark.readStream .format(\"delta\")\\n  .option(\"ignoreChanges\" , \"true\")\\n  .load(\"/mnt/delta/user_events\" )\\nIf you update a user_email with the UPDATE statement, the file containing the\\nuser_email in question is rewritten. When you use ignoreChanges, the new record is\\npropagated downstream with all other unchanged records that were in the same file.\\nY our logic should be able to handle these incoming duplicate records.\\n100 | Chapter 3: Continuous Applications with Delta LakeSpecify initial position\\nThis feature is only available on Databricks Runtime 7.3 LTS and\\nabove.\\nInitial position is where you want the stream to start with as a default position. Defin‐\\ning this position, allows you to specify a streaming source without processing the\\nentire table. Since Delta saves each record in the form of versions along with the\\ntimestamps, you can specify initial position by using either of these two values.\\n•startingVersion : startingVersion  allows you to specify the Delta version of\\nthe table you want to start from. All table changes starting from this version\\n(inclusive) will be read by the streaming source.\\n•startingTimestamp : This allows you to specify the timestamp to start from. All\\ntable changes committed at or after the timestamp (inclusive) will be read by the\\nstreaming source.\\nY ou can retrieve both the above information using DESCRIBE command:\\nDESCRIBE HISTORY customer_t1;\\nThe following results are an abridged version of the table, focusing only on the ver‐\\nsion, timestamp, and operation columns.\\n-- Abridged Results\\n+-------+-------------------+---------+\\n|version|timestamp|operation|\\n+-------+-------------------+---------+\\n|     19|2020-10-30 18:15:03|    WRITE|\\n|     18|2020-10-30 18:10:47|   DELETE|\\n|     17|2020-10-30 08:41:47|    WRITE|\\nThe operations take effect only when starting a new streaming query. If a streaming\\nquery has started and the progress has been recorded in its checkpoint, these options\\nare ignored.\\nExample\\nFor example, suppose you have a table user_events. If you want to read changes since\\nversion 5, use:\\n%Scala\\nspark.readStream .format(\"delta\")\\n  .option(\"startingVersion\" , \"5\")\\n  .load(\"/mnt/delta/user_events\" )\\nMake All Your Streams Come True | 101If you want to read changes since 2018-10-18, use:\\n%Scala\\nspark.readStream .format(\"delta\")\\n  .option(\"startingTimestamp\" , \"2018-10-18\" )\\n  .load(\"/mnt/delta/user_events\" )\\nDelta Table as a Sink\\nY ou can also write data into a Delta table using Structured Streaming. The transaction\\nlog enables Delta Lake to guarantee exactly-once processing, even when there are\\nother streams or batch queries running concurrently against the table.\\nAppend mode\\nBy default, streams run in append mode, which adds new records to the table.\\nY ou can use the path method:\\n%Scala\\nevents.writeStream\\n  .format(\"delta\")\\n  .outputMode (\"append\" )\\n  .option(\"checkpointLocation\" , \"/delta/events/_checkpoints/etl-from-json\" )\\n  .start(\"/delta/events\" )\\nor the table method:\\n%Scala\\nevents.writeStream\\n  .format(\"delta\")\\n  .outputMode (\"append\" )\\n  .option(\"checkpointLocation\" , \"/delta/events/_checkpoints/etl-from-json\" )\\n  .table(\"events\" )\\nComplete mode\\nY ou can also use Structured Streaming to replace the entire table with every batch.\\nOne example use case is to compute a summary using aggregation:\\n%Scala\\nspark.readStream\\n  .format(\"delta\")\\n  .load(\"/mnt/delta/events\" )\\n  .groupBy(\"customerId\" )\\n  .count()\\n  .writeStream\\n  .format(\"delta\")\\n  .outputMode (\"complete\" )\\n  .option(\"checkpointLocation\" , \"/mnt/delta/eventsByCustomer/_checkpoints/streaming-\\nagg\")\\n  .start(\"/mnt/delta/eventsByCustomer\" )\\n102 | Chapter 3: Continuous Applications with Delta LakeThe preceding example continuously updates a table that contains the aggregate\\nnumber of events by customer.\\nFor applications with more lenient latency requirements, you can save computing\\nresources with one-time triggers. Use these to update summary aggregation tables on\\na given schedule, processing only new data that has arrived since the last update.\\nAs a summary, we covered the evolution of streaming from Apache Storm to spark\\nstreaming to structured streaming and looked at the advantages of using Delta tables\\nfor the unified pipeline for batch and streaming source and sink. In the next chapter\\nMedallion Architecture, we will cover the design of Delta Lake architecture and some\\nof the evolving features like Delta Live tables.\\nAppendix\\nFor more examples of streaming with Delta, please refer to the following notebooks:\\n•Using SQL Analytics to Query Y our Data Lake With Delta Lake - Databricks\\n•Using SQL to Query Y our Data Lake With Delta Lake - Databricks\\nMake All Your Streams Come True | 103About the Authors\\nDenny Lee  is a Staff Developer Advocate at Databricks. He is a hands-on distributed\\nsystems and data sciences engineer with extensive experience developing internet-\\nscale infrastructure, data platforms, and predictive analytics systems for both on-\\npremise and cloud environments. He also has a Masters of Biomedical Informatics\\nfrom Oregon Health and Sciences University and has architected and implemented\\npowerful data solutions for enterprise Healthcare customers. His current technical\\nfocuses include Distributed Systems, Apache Spark, Deep Learning, Machine Learn‐\\ning, and Genomics.\\nTathagata Das  (TD) is a Staff Software Engineer at Databricks, an Apache Spark\\ncommitter and a member of the Apache Spark Project Management Committee\\n(PMC). He is one of the original developers of Apache Spark, the lead developer of\\nSpark Streaming (DStreams) and is currently one of the core developers of Structured\\nStreaming and Delta Lake. Tathagata holds a MS in computer science from UC\\nBerkeley.\\nVini Jaiswal  is a Senior Developer Advocate at Databricks, where she helps data prac‐\\ntitioners to be successful building on Databricks and open source technologies like\\nApache Spark, Delta and MLflow. She has extensive experience working with the\\nUnicorns, Digital Natives and some of the Fortune 500 companies helping with suc‐\\ncessful implementation of Data and AI use cases in production at scale for on-\\npremise and cloud deployments. Vini also worked as the Data Science Engineering\\nLead under Citi’s Enterprise Operations & Technology group and interned as a Data\\nAnalyst at Southwest Airlines. She holds an MS in Information Technology and Man‐\\nagement from University of Texas at Dallas.'"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "raw_text"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "1b5cd83f-d8d7-4690-9736-42274f89e351",
   "metadata": {},
   "outputs": [],
   "source": [
    "text_splitter = CharacterTextSplitter(\n",
    "    separator = \"\\n\",\n",
    "    chunk_size = 800,\n",
    "    chunk_overlap  = 200,\n",
    "    length_function = len,\n",
    ")\n",
    "texts = text_splitter.split_text(raw_text)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "8f2a2557-3b7f-4db1-b405-54c863810eb4",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "279"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "len(texts)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "bea595ff-e65f-4edd-9c80-1ab31052f606",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "C:\\Users\\ajayk\\anaconda3\\lib\\site-packages\\langchain_core\\_api\\deprecation.py:119: LangChainDeprecationWarning: The class `OpenAIEmbeddings` was deprecated in LangChain 0.0.9 and will be removed in 0.2.0. An updated version of the class exists in the langchain-openai package and should be used instead. To use it run `pip install -U langchain-openai` and import as `from langchain_openai import OpenAIEmbeddings`.\n",
      "  warn_deprecated(\n"
     ]
    }
   ],
   "source": [
    "embeddings = OpenAIEmbeddings()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "id": "6d832fb7-c375-4685-bdee-64c899dca44b",
   "metadata": {},
   "outputs": [],
   "source": [
    "document_search = Chroma.from_texts(texts, embeddings)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "729c1ab6-b16c-4998-9d5e-bfe60bf40556",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<langchain_community.vectorstores.chroma.Chroma at 0x1dfaec22100>"
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "document_search"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "223b8096-d3f8-4d6c-9b59-1fe373865d6a",
   "metadata": {},
   "outputs": [],
   "source": [
    "from langchain.chains.question_answering import load_qa_chain\n",
    "from langchain.llms import OpenAI\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "id": "2c1f25c0-fc8c-4d42-92f9-583dccabd49e",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "C:\\Users\\ajayk\\anaconda3\\lib\\site-packages\\langchain_core\\_api\\deprecation.py:119: LangChainDeprecationWarning: The class `OpenAI` was deprecated in LangChain 0.0.10 and will be removed in 0.2.0. An updated version of the class exists in the langchain-openai package and should be used instead. To use it run `pip install -U langchain-openai` and import as `from langchain_openai import OpenAI`.\n",
      "  warn_deprecated(\n"
     ]
    }
   ],
   "source": [
    "chain = load_qa_chain(OpenAI(), chain_type=\"stuff\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "id": "ce41e460-928c-428d-b830-ee1541b77515",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'\\nDelta Lake is a storage layer that provides ACID transactions, scalable metadata handling, and unified streaming and batch data processing in a single solution. It is built on top of Apache Spark and is designed to address common data lake challenges such as data reliability, scalability, and performance.\\n\\nAt its core, Delta Lake is a transaction log that stores all the changes made to the data lake. This log is structured in a way that allows Delta Lake to implement the principle of atomicity, which means that either all the changes in a transaction are applied or none of them are. This ensures data consistency and reliability, even in the event of failures.\\n\\nDelta Lake uses a single source of truth approach, which means that all changes to the data lake are recorded in the transaction log, and this log is the only source of truth for the data lake. This allows Delta Lake to easily compute the state of each table, using the transaction log to catch up from the most recent checkpoint. This ensures that the data lake is always up to date and consistent.\\n\\nOne of the key features of Delta Lake is its support for concurrent reads and writes. This is achieved through optimistic concurrency control, where multiple users can read and write to the same table at the same time without blocking each other. Delta Lake also'"
      ]
     },
     "execution_count": 41,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "query = \"what is a delta lake explain in detail\"\n",
    "docs = document_search.similarity_search(query)\n",
    "chain.run(input_documents=docs, question=query)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a3f22c20-b5d4-4747-8e5b-b3ba79780009",
   "metadata": {},
   "source": [
    "### RAG using llama2"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "c65f2bc9-548d-43fa-ba85-19641c54c112",
   "metadata": {},
   "outputs": [],
   "source": [
    "embedding1 = OllamaEmbeddings(model_name=\"llama2\")\n",
    "#embedding1=OllamaEmbeddings(model=\"nomic-embed-text\",show_progress=True),"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "id": "1cf8550f-d6dc-4f91-bd21-8740e89dc353",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "OllamaEmbeddings(base_url='http://localhost:11434', model='llama2', embed_instruction='passage: ', query_instruction='query: ', mirostat=None, mirostat_eta=None, mirostat_tau=None, num_ctx=None, num_gpu=None, num_thread=None, repeat_last_n=None, repeat_penalty=None, temperature=None, stop=None, tfs_z=None, top_k=None, top_p=None, show_progress=False, headers=None, model_kwargs=None)"
      ]
     },
     "execution_count": 25,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "embedding1"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c103ba91",
   "metadata": {},
   "source": [
    "## MultiQueryRetriever\n",
    "The MultiQueryRetriever is likely used to enhance the retrieval process by enabling multiple queries to be executed, thereby improving the accuracy and relevance of the retrieved information. This can be particularly useful in applications such as:\n",
    "\n",
    "1. Question Answering Systems\n",
    "2. Chatbots\n",
    "3. Search Engines"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "4ab1e457-1370-4ce5-a6fe-ac071a623911",
   "metadata": {},
   "outputs": [],
   "source": [
    "from langchain.prompts import ChatPromptTemplate, PromptTemplate\n",
    "from langchain_core.output_parsers import StrOutputParser\n",
    "from langchain_community.chat_models import ChatOllama\n",
    "from langchain_core.runnables import RunnablePassthrough\n",
    "from langchain.retrievers.multi_query import MultiQueryRetriever"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "id": "6e86f153-703c-4092-8800-d899dd661d62",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\n",
      "OllamaEmbeddings:   0%|                                                                        | 0/279 [00:00<?, ?it/s]\u001b[A\n",
      "OllamaEmbeddings:   0%|▏                                                               | 1/279 [00:04<19:09,  4.14s/it]\u001b[A\n",
      "OllamaEmbeddings:   1%|▍                                                               | 2/279 [00:06<15:10,  3.29s/it]\u001b[A\n",
      "OllamaEmbeddings:   1%|▋                                                               | 3/279 [00:09<13:52,  3.02s/it]\u001b[A\n",
      "OllamaEmbeddings:   1%|▉                                                               | 4/279 [00:13<14:41,  3.20s/it]\u001b[A\n",
      "OllamaEmbeddings:   2%|█▏                                                              | 5/279 [00:16<14:48,  3.24s/it]\u001b[A\n",
      "OllamaEmbeddings:   2%|█▍                                                              | 6/279 [00:18<13:39,  3.00s/it]\u001b[A\n",
      "OllamaEmbeddings:   3%|█▌                                                              | 7/279 [00:21<12:46,  2.82s/it]\u001b[A\n",
      "OllamaEmbeddings:   3%|█▊                                                              | 8/279 [00:23<12:14,  2.71s/it]\u001b[A\n",
      "OllamaEmbeddings:   3%|██                                                              | 9/279 [00:26<12:26,  2.76s/it]\u001b[A\n",
      "OllamaEmbeddings:   4%|██▎                                                            | 10/279 [00:29<11:51,  2.65s/it]\u001b[A\n",
      "OllamaEmbeddings:   4%|██▍                                                            | 11/279 [00:31<11:58,  2.68s/it]\u001b[A\n",
      "OllamaEmbeddings:   4%|██▋                                                            | 12/279 [00:34<12:19,  2.77s/it]\u001b[A\n",
      "OllamaEmbeddings:   5%|██▉                                                            | 13/279 [00:37<12:26,  2.81s/it]\u001b[A\n",
      "OllamaEmbeddings:   5%|███▏                                                           | 14/279 [00:41<13:24,  3.04s/it]\u001b[A\n",
      "OllamaEmbeddings:   5%|███▍                                                           | 15/279 [00:44<14:01,  3.19s/it]\u001b[A\n",
      "OllamaEmbeddings:   6%|███▌                                                           | 16/279 [00:48<14:25,  3.29s/it]\u001b[A\n",
      "OllamaEmbeddings:   6%|███▊                                                           | 17/279 [00:51<14:36,  3.35s/it]\u001b[A\n",
      "OllamaEmbeddings:   6%|████                                                           | 18/279 [00:55<14:45,  3.39s/it]\u001b[A\n",
      "OllamaEmbeddings:   7%|████▎                                                          | 19/279 [00:58<14:50,  3.42s/it]\u001b[A\n",
      "OllamaEmbeddings:   7%|████▌                                                          | 20/279 [01:02<14:50,  3.44s/it]\u001b[A\n",
      "OllamaEmbeddings:   8%|████▋                                                          | 21/279 [01:05<14:58,  3.48s/it]\u001b[A\n",
      "OllamaEmbeddings:   8%|████▉                                                          | 22/279 [01:09<15:11,  3.55s/it]\u001b[A\n",
      "OllamaEmbeddings:   8%|█████▏                                                         | 23/279 [01:13<15:12,  3.57s/it]\u001b[A\n",
      "OllamaEmbeddings:   9%|█████▍                                                         | 24/279 [01:17<15:44,  3.70s/it]\u001b[A\n",
      "OllamaEmbeddings:   9%|█████▋                                                         | 25/279 [01:21<16:07,  3.81s/it]\u001b[A\n",
      "OllamaEmbeddings:   9%|█████▊                                                         | 26/279 [01:25<16:10,  3.83s/it]\u001b[A\n",
      "OllamaEmbeddings:  10%|██████                                                         | 27/279 [01:28<15:57,  3.80s/it]\u001b[A\n",
      "OllamaEmbeddings:  10%|██████▎                                                        | 28/279 [01:32<15:57,  3.82s/it]\u001b[A\n",
      "OllamaEmbeddings:  10%|██████▌                                                        | 29/279 [01:36<16:07,  3.87s/it]\u001b[A\n",
      "OllamaEmbeddings:  11%|██████▊                                                        | 30/279 [01:40<16:23,  3.95s/it]\u001b[A\n",
      "OllamaEmbeddings:  11%|███████                                                        | 31/279 [01:45<16:54,  4.09s/it]\u001b[A\n",
      "OllamaEmbeddings:  11%|███████▏                                                       | 32/279 [01:49<17:05,  4.15s/it]\u001b[A\n",
      "OllamaEmbeddings:  12%|███████▍                                                       | 33/279 [01:53<16:42,  4.07s/it]\u001b[A\n",
      "OllamaEmbeddings:  12%|███████▋                                                       | 34/279 [01:57<16:36,  4.07s/it]\u001b[A\n",
      "OllamaEmbeddings:  13%|███████▉                                                       | 35/279 [02:01<16:07,  3.97s/it]\u001b[A\n",
      "OllamaEmbeddings:  13%|████████▏                                                      | 36/279 [02:04<15:33,  3.84s/it]\u001b[A\n",
      "OllamaEmbeddings:  13%|████████▎                                                      | 37/279 [02:08<15:02,  3.73s/it]\u001b[A\n",
      "OllamaEmbeddings:  14%|████████▌                                                      | 38/279 [02:11<14:55,  3.71s/it]\u001b[A\n",
      "OllamaEmbeddings:  14%|████████▊                                                      | 39/279 [02:15<14:51,  3.71s/it]\u001b[A\n",
      "OllamaEmbeddings:  14%|█████████                                                      | 40/279 [02:19<14:36,  3.67s/it]\u001b[A\n",
      "OllamaEmbeddings:  15%|█████████▎                                                     | 41/279 [02:22<14:41,  3.70s/it]\u001b[A\n",
      "OllamaEmbeddings:  15%|█████████▍                                                     | 42/279 [02:26<14:31,  3.68s/it]\u001b[A\n",
      "OllamaEmbeddings:  15%|█████████▋                                                     | 43/279 [02:30<14:37,  3.72s/it]\u001b[A\n",
      "OllamaEmbeddings:  16%|█████████▉                                                     | 44/279 [02:34<14:29,  3.70s/it]\u001b[A\n",
      "OllamaEmbeddings:  16%|██████████▏                                                    | 45/279 [02:37<14:15,  3.66s/it]\u001b[A\n",
      "OllamaEmbeddings:  16%|██████████▍                                                    | 46/279 [02:41<14:11,  3.65s/it]\u001b[A\n",
      "OllamaEmbeddings:  17%|██████████▌                                                    | 47/279 [02:44<14:00,  3.62s/it]\u001b[A\n",
      "OllamaEmbeddings:  17%|██████████▊                                                    | 48/279 [02:48<13:44,  3.57s/it]\u001b[A\n",
      "OllamaEmbeddings:  18%|███████████                                                    | 49/279 [02:51<13:23,  3.49s/it]\u001b[A\n",
      "OllamaEmbeddings:  18%|███████████▎                                                   | 50/279 [02:55<13:30,  3.54s/it]\u001b[A\n",
      "OllamaEmbeddings:  18%|███████████▌                                                   | 51/279 [02:58<13:36,  3.58s/it]\u001b[A\n",
      "OllamaEmbeddings:  19%|███████████▋                                                   | 52/279 [03:02<13:40,  3.62s/it]\u001b[A\n",
      "OllamaEmbeddings:  19%|███████████▉                                                   | 53/279 [03:06<13:42,  3.64s/it]\u001b[A\n",
      "OllamaEmbeddings:  19%|████████████▏                                                  | 54/279 [03:10<13:53,  3.70s/it]\u001b[A\n",
      "OllamaEmbeddings:  20%|████████████▍                                                  | 55/279 [03:13<13:54,  3.72s/it]\u001b[A\n",
      "OllamaEmbeddings:  20%|████████████▋                                                  | 56/279 [03:17<14:08,  3.81s/it]\u001b[A\n",
      "OllamaEmbeddings:  20%|████████████▊                                                  | 57/279 [03:22<14:53,  4.03s/it]\u001b[A\n",
      "OllamaEmbeddings:  21%|█████████████                                                  | 58/279 [03:27<15:52,  4.31s/it]\u001b[A\n",
      "OllamaEmbeddings:  21%|█████████████▎                                                 | 59/279 [03:31<15:34,  4.25s/it]\u001b[A\n",
      "OllamaEmbeddings:  22%|█████████████▌                                                 | 60/279 [03:36<15:54,  4.36s/it]\u001b[A\n",
      "OllamaEmbeddings:  22%|█████████████▊                                                 | 61/279 [03:39<15:01,  4.13s/it]\u001b[A\n",
      "OllamaEmbeddings:  22%|██████████████                                                 | 62/279 [03:43<14:19,  3.96s/it]\u001b[A\n",
      "OllamaEmbeddings:  23%|██████████████▏                                                | 63/279 [03:47<14:20,  3.99s/it]\u001b[A\n",
      "OllamaEmbeddings:  23%|██████████████▍                                                | 64/279 [03:50<13:46,  3.84s/it]\u001b[A\n",
      "OllamaEmbeddings:  23%|██████████████▋                                                | 65/279 [03:54<13:39,  3.83s/it]\u001b[A\n",
      "OllamaEmbeddings:  24%|██████████████▉                                                | 66/279 [03:58<13:36,  3.83s/it]\u001b[A\n",
      "OllamaEmbeddings:  24%|███████████████▏                                               | 67/279 [04:02<13:27,  3.81s/it]\u001b[A\n",
      "OllamaEmbeddings:  24%|███████████████▎                                               | 68/279 [04:06<13:27,  3.83s/it]\u001b[A\n",
      "OllamaEmbeddings:  25%|███████████████▌                                               | 69/279 [04:10<13:57,  3.99s/it]\u001b[A\n",
      "OllamaEmbeddings:  25%|███████████████▊                                               | 70/279 [04:15<14:54,  4.28s/it]\u001b[A\n",
      "OllamaEmbeddings:  25%|████████████████                                               | 71/279 [04:19<14:43,  4.25s/it]\u001b[A\n",
      "OllamaEmbeddings:  26%|████████████████▎                                              | 72/279 [04:23<13:54,  4.03s/it]\u001b[A\n",
      "OllamaEmbeddings:  26%|████████████████▍                                              | 73/279 [04:26<13:10,  3.84s/it]\u001b[A\n",
      "OllamaEmbeddings:  27%|████████████████▋                                              | 74/279 [04:30<12:57,  3.80s/it]\u001b[A\n",
      "OllamaEmbeddings:  27%|████████████████▉                                              | 75/279 [04:34<12:56,  3.81s/it]\u001b[A\n",
      "OllamaEmbeddings:  27%|█████████████████▏                                             | 76/279 [04:37<12:40,  3.74s/it]\u001b[A\n",
      "OllamaEmbeddings:  28%|█████████████████▍                                             | 77/279 [04:41<12:28,  3.70s/it]\u001b[A\n",
      "OllamaEmbeddings:  28%|█████████████████▌                                             | 78/279 [04:45<12:32,  3.74s/it]\u001b[A\n",
      "OllamaEmbeddings:  28%|█████████████████▊                                             | 79/279 [04:48<12:34,  3.77s/it]\u001b[A\n",
      "OllamaEmbeddings:  29%|██████████████████                                             | 80/279 [04:52<12:27,  3.75s/it]\u001b[A\n",
      "OllamaEmbeddings:  29%|██████████████████▎                                            | 81/279 [04:56<12:15,  3.71s/it]\u001b[A\n",
      "OllamaEmbeddings:  29%|██████████████████▌                                            | 82/279 [04:59<12:01,  3.66s/it]\u001b[A\n",
      "OllamaEmbeddings:  30%|██████████████████▋                                            | 83/279 [05:03<11:46,  3.60s/it]\u001b[A\n",
      "OllamaEmbeddings:  30%|██████████████████▉                                            | 84/279 [05:06<11:38,  3.58s/it]\u001b[A\n",
      "OllamaEmbeddings:  30%|███████████████████▏                                           | 85/279 [05:10<11:38,  3.60s/it]\u001b[A\n",
      "OllamaEmbeddings:  31%|███████████████████▍                                           | 86/279 [05:14<11:42,  3.64s/it]\u001b[A\n",
      "OllamaEmbeddings:  31%|███████████████████▋                                           | 87/279 [05:17<11:39,  3.64s/it]\u001b[A\n",
      "OllamaEmbeddings:  32%|███████████████████▊                                           | 88/279 [05:21<11:32,  3.63s/it]\u001b[A\n",
      "OllamaEmbeddings:  32%|████████████████████                                           | 89/279 [05:25<11:37,  3.67s/it]\u001b[A\n",
      "OllamaEmbeddings:  32%|████████████████████▎                                          | 90/279 [05:29<11:44,  3.73s/it]\u001b[A\n",
      "OllamaEmbeddings:  33%|████████████████████▌                                          | 91/279 [05:33<12:06,  3.86s/it]\u001b[A\n",
      "OllamaEmbeddings:  33%|████████████████████▊                                          | 92/279 [05:36<11:48,  3.79s/it]\u001b[A\n",
      "OllamaEmbeddings:  33%|█████████████████████                                          | 93/279 [05:40<11:41,  3.77s/it]\u001b[A\n",
      "OllamaEmbeddings:  34%|█████████████████████▏                                         | 94/279 [05:44<11:37,  3.77s/it]\u001b[A\n",
      "OllamaEmbeddings:  34%|█████████████████████▍                                         | 95/279 [05:48<11:39,  3.80s/it]\u001b[A\n",
      "OllamaEmbeddings:  34%|█████████████████████▋                                         | 96/279 [05:51<11:34,  3.79s/it]\u001b[A\n",
      "OllamaEmbeddings:  35%|█████████████████████▉                                         | 97/279 [05:55<11:21,  3.74s/it]\u001b[A\n",
      "OllamaEmbeddings:  35%|██████████████████████▏                                        | 98/279 [05:59<11:13,  3.72s/it]\u001b[A\n",
      "OllamaEmbeddings:  35%|██████████████████████▎                                        | 99/279 [06:02<11:02,  3.68s/it]\u001b[A\n",
      "OllamaEmbeddings:  36%|██████████████████████▏                                       | 100/279 [06:06<10:58,  3.68s/it]\u001b[A\n",
      "OllamaEmbeddings:  36%|██████████████████████▍                                       | 101/279 [06:10<11:21,  3.83s/it]\u001b[A\n",
      "OllamaEmbeddings:  37%|██████████████████████▋                                       | 102/279 [06:14<11:22,  3.85s/it]\u001b[A\n",
      "OllamaEmbeddings:  37%|██████████████████████▉                                       | 103/279 [06:19<12:07,  4.13s/it]\u001b[A\n",
      "OllamaEmbeddings:  37%|███████████████████████                                       | 104/279 [06:23<12:04,  4.14s/it]\u001b[A\n",
      "OllamaEmbeddings:  38%|███████████████████████▎                                      | 105/279 [06:27<12:04,  4.16s/it]\u001b[A\n",
      "OllamaEmbeddings:  38%|███████████████████████▌                                      | 106/279 [06:31<11:56,  4.14s/it]\u001b[A\n",
      "OllamaEmbeddings:  38%|███████████████████████▊                                      | 107/279 [06:35<11:48,  4.12s/it]\u001b[A\n",
      "OllamaEmbeddings:  39%|████████████████████████                                      | 108/279 [06:40<11:53,  4.17s/it]\u001b[A\n",
      "OllamaEmbeddings:  39%|████████████████████████▏                                     | 109/279 [06:44<11:55,  4.21s/it]\u001b[A\n",
      "OllamaEmbeddings:  39%|████████████████████████▍                                     | 110/279 [06:48<11:26,  4.06s/it]\u001b[A\n",
      "OllamaEmbeddings:  40%|████████████████████████▋                                     | 111/279 [06:51<11:03,  3.95s/it]\u001b[A\n",
      "OllamaEmbeddings:  40%|████████████████████████▉                                     | 112/279 [06:56<11:10,  4.01s/it]\u001b[A\n",
      "OllamaEmbeddings:  41%|█████████████████████████                                     | 113/279 [07:00<11:08,  4.03s/it]\u001b[A\n",
      "OllamaEmbeddings:  41%|█████████████████████████▎                                    | 114/279 [07:03<10:50,  3.94s/it]\u001b[A\n",
      "OllamaEmbeddings:  41%|█████████████████████████▌                                    | 115/279 [07:07<10:31,  3.85s/it]\u001b[A\n",
      "OllamaEmbeddings:  42%|█████████████████████████▊                                    | 116/279 [07:11<10:23,  3.82s/it]\u001b[A\n",
      "OllamaEmbeddings:  42%|██████████████████████████                                    | 117/279 [07:15<10:14,  3.79s/it]\u001b[A\n",
      "OllamaEmbeddings:  42%|██████████████████████████▏                                   | 118/279 [07:18<09:57,  3.71s/it]\u001b[A\n",
      "OllamaEmbeddings:  43%|██████████████████████████▍                                   | 119/279 [07:22<10:05,  3.78s/it]\u001b[A\n",
      "OllamaEmbeddings:  43%|██████████████████████████▋                                   | 120/279 [07:26<10:08,  3.83s/it]\u001b[A\n",
      "OllamaEmbeddings:  43%|██████████████████████████▉                                   | 121/279 [07:30<10:06,  3.84s/it]\u001b[A\n",
      "OllamaEmbeddings:  44%|███████████████████████████                                   | 122/279 [07:33<09:50,  3.76s/it]\u001b[A\n",
      "OllamaEmbeddings:  44%|███████████████████████████▎                                  | 123/279 [07:37<09:40,  3.72s/it]\u001b[A\n",
      "OllamaEmbeddings:  44%|███████████████████████████▌                                  | 124/279 [07:41<09:44,  3.77s/it]\u001b[A\n",
      "OllamaEmbeddings:  45%|███████████████████████████▊                                  | 125/279 [07:45<09:39,  3.76s/it]\u001b[A\n",
      "OllamaEmbeddings:  45%|████████████████████████████                                  | 126/279 [07:48<09:27,  3.71s/it]\u001b[A\n",
      "OllamaEmbeddings:  46%|████████████████████████████▏                                 | 127/279 [07:52<09:23,  3.70s/it]\u001b[A\n",
      "OllamaEmbeddings:  46%|████████████████████████████▍                                 | 128/279 [07:55<09:13,  3.67s/it]\u001b[A\n",
      "OllamaEmbeddings:  46%|████████████████████████████▋                                 | 129/279 [07:59<08:57,  3.58s/it]\u001b[A\n",
      "OllamaEmbeddings:  47%|████████████████████████████▉                                 | 130/279 [08:02<08:52,  3.57s/it]\u001b[A\n",
      "OllamaEmbeddings:  47%|█████████████████████████████                                 | 131/279 [08:06<08:54,  3.61s/it]\u001b[A\n",
      "OllamaEmbeddings:  47%|█████████████████████████████▎                                | 132/279 [08:10<08:56,  3.65s/it]\u001b[A\n",
      "OllamaEmbeddings:  48%|█████████████████████████████▌                                | 133/279 [08:13<08:43,  3.59s/it]\u001b[A\n",
      "OllamaEmbeddings:  48%|█████████████████████████████▊                                | 134/279 [08:17<08:41,  3.60s/it]\u001b[A\n",
      "OllamaEmbeddings:  48%|██████████████████████████████                                | 135/279 [08:21<08:49,  3.68s/it]\u001b[A\n",
      "OllamaEmbeddings:  49%|██████████████████████████████▏                               | 136/279 [08:25<09:12,  3.87s/it]\u001b[A\n",
      "OllamaEmbeddings:  49%|██████████████████████████████▍                               | 137/279 [08:31<10:34,  4.47s/it]\u001b[A\n",
      "OllamaEmbeddings:  49%|██████████████████████████████▋                               | 138/279 [08:35<10:20,  4.40s/it]\u001b[A\n",
      "OllamaEmbeddings:  50%|██████████████████████████████▉                               | 139/279 [08:39<09:48,  4.21s/it]\u001b[A\n",
      "OllamaEmbeddings:  50%|███████████████████████████████                               | 140/279 [08:44<10:00,  4.32s/it]\u001b[A\n",
      "OllamaEmbeddings:  51%|███████████████████████████████▎                              | 141/279 [08:49<10:55,  4.75s/it]\u001b[A\n",
      "OllamaEmbeddings:  51%|███████████████████████████████▌                              | 142/279 [08:53<10:20,  4.53s/it]\u001b[A\n",
      "OllamaEmbeddings:  51%|███████████████████████████████▊                              | 143/279 [08:58<10:08,  4.47s/it]\u001b[A\n",
      "OllamaEmbeddings:  52%|████████████████████████████████                              | 144/279 [09:02<09:38,  4.28s/it]\u001b[A\n",
      "OllamaEmbeddings:  52%|████████████████████████████████▏                             | 145/279 [09:07<10:30,  4.70s/it]\u001b[A\n",
      "OllamaEmbeddings:  52%|████████████████████████████████▍                             | 146/279 [09:13<11:17,  5.09s/it]\u001b[A\n",
      "OllamaEmbeddings:  53%|████████████████████████████████▋                             | 147/279 [09:18<10:41,  4.86s/it]\u001b[A\n",
      "OllamaEmbeddings:  53%|████████████████████████████████▉                             | 148/279 [09:21<09:52,  4.52s/it]\u001b[A\n",
      "OllamaEmbeddings:  53%|█████████████████████████████████                             | 149/279 [09:25<09:21,  4.32s/it]\u001b[A\n",
      "OllamaEmbeddings:  54%|█████████████████████████████████▎                            | 150/279 [09:29<09:13,  4.29s/it]\u001b[A\n",
      "OllamaEmbeddings:  54%|█████████████████████████████████▌                            | 151/279 [09:35<10:08,  4.75s/it]\u001b[A\n",
      "OllamaEmbeddings:  54%|█████████████████████████████████▊                            | 152/279 [09:41<10:48,  5.10s/it]\u001b[A\n",
      "OllamaEmbeddings:  55%|██████████████████████████████████                            | 153/279 [09:45<10:15,  4.89s/it]\u001b[A\n",
      "OllamaEmbeddings:  55%|██████████████████████████████████▏                           | 154/279 [09:50<09:51,  4.73s/it]\u001b[A\n",
      "OllamaEmbeddings:  56%|██████████████████████████████████▍                           | 155/279 [09:54<09:39,  4.68s/it]\u001b[A\n",
      "OllamaEmbeddings:  56%|██████████████████████████████████▋                           | 156/279 [10:01<10:34,  5.16s/it]\u001b[A\n",
      "OllamaEmbeddings:  56%|██████████████████████████████████▉                           | 157/279 [10:06<10:44,  5.28s/it]\u001b[A\n",
      "OllamaEmbeddings:  57%|███████████████████████████████████                           | 158/279 [10:12<10:42,  5.31s/it]\u001b[A\n",
      "OllamaEmbeddings:  57%|███████████████████████████████████▎                          | 159/279 [10:18<11:04,  5.54s/it]\u001b[A\n",
      "OllamaEmbeddings:  57%|███████████████████████████████████▌                          | 160/279 [10:22<10:03,  5.07s/it]\u001b[A\n",
      "OllamaEmbeddings:  58%|███████████████████████████████████▊                          | 161/279 [10:26<09:47,  4.98s/it]\u001b[A\n",
      "OllamaEmbeddings:  58%|████████████████████████████████████                          | 162/279 [10:32<09:46,  5.02s/it]\u001b[A\n",
      "OllamaEmbeddings:  58%|████████████████████████████████████▏                         | 163/279 [10:35<08:59,  4.65s/it]\u001b[A\n",
      "OllamaEmbeddings:  59%|████████████████████████████████████▍                         | 164/279 [10:40<09:12,  4.80s/it]\u001b[A\n",
      "OllamaEmbeddings:  59%|████████████████████████████████████▋                         | 165/279 [10:47<10:19,  5.44s/it]\u001b[A\n",
      "OllamaEmbeddings:  59%|████████████████████████████████████▉                         | 166/279 [10:52<09:38,  5.12s/it]\u001b[A\n",
      "OllamaEmbeddings:  60%|█████████████████████████████████████                         | 167/279 [10:57<09:43,  5.21s/it]\u001b[A\n",
      "OllamaEmbeddings:  60%|█████████████████████████████████████▎                        | 168/279 [11:02<09:36,  5.19s/it]\u001b[A\n",
      "OllamaEmbeddings:  61%|█████████████████████████████████████▌                        | 169/279 [11:06<08:54,  4.86s/it]\u001b[A\n",
      "OllamaEmbeddings:  61%|█████████████████████████████████████▊                        | 170/279 [11:10<08:17,  4.56s/it]\u001b[A\n",
      "OllamaEmbeddings:  61%|██████████████████████████████████████                        | 171/279 [11:14<07:50,  4.36s/it]\u001b[A\n",
      "OllamaEmbeddings:  62%|██████████████████████████████████████▏                       | 172/279 [11:19<08:11,  4.60s/it]\u001b[A\n",
      "OllamaEmbeddings:  62%|██████████████████████████████████████▍                       | 173/279 [11:24<08:19,  4.72s/it]\u001b[A\n",
      "OllamaEmbeddings:  62%|██████████████████████████████████████▋                       | 174/279 [11:30<08:54,  5.09s/it]\u001b[A\n",
      "OllamaEmbeddings:  63%|██████████████████████████████████████▉                       | 175/279 [11:35<08:37,  4.98s/it]\u001b[A\n",
      "OllamaEmbeddings:  63%|███████████████████████████████████████                       | 176/279 [11:39<08:12,  4.78s/it]\u001b[A\n",
      "OllamaEmbeddings:  63%|███████████████████████████████████████▎                      | 177/279 [11:44<08:02,  4.73s/it]\u001b[A\n",
      "OllamaEmbeddings:  64%|███████████████████████████████████████▌                      | 178/279 [11:50<08:29,  5.04s/it]\u001b[A\n",
      "OllamaEmbeddings:  64%|███████████████████████████████████████▊                      | 179/279 [11:54<08:15,  4.95s/it]\u001b[A\n",
      "OllamaEmbeddings:  65%|████████████████████████████████████████                      | 180/279 [12:01<08:43,  5.29s/it]\u001b[A\n",
      "OllamaEmbeddings:  65%|████████████████████████████████████████▏                     | 181/279 [12:07<09:28,  5.80s/it]\u001b[A\n",
      "OllamaEmbeddings:  65%|████████████████████████████████████████▍                     | 182/279 [12:14<09:40,  5.99s/it]\u001b[A\n",
      "OllamaEmbeddings:  66%|████████████████████████████████████████▋                     | 183/279 [12:19<09:13,  5.76s/it]\u001b[A\n",
      "OllamaEmbeddings:  66%|████████████████████████████████████████▉                     | 184/279 [12:25<08:56,  5.65s/it]\u001b[A\n",
      "OllamaEmbeddings:  66%|█████████████████████████████████████████                     | 185/279 [12:31<09:00,  5.75s/it]\u001b[A\n",
      "OllamaEmbeddings:  67%|█████████████████████████████████████████▎                    | 186/279 [12:37<09:04,  5.85s/it]\u001b[A\n",
      "OllamaEmbeddings:  67%|█████████████████████████████████████████▌                    | 187/279 [12:41<08:18,  5.42s/it]\u001b[A\n",
      "OllamaEmbeddings:  67%|█████████████████████████████████████████▊                    | 188/279 [12:45<07:23,  4.88s/it]\u001b[A\n",
      "OllamaEmbeddings:  68%|██████████████████████████████████████████                    | 189/279 [12:48<06:45,  4.50s/it]\u001b[A\n",
      "OllamaEmbeddings:  68%|██████████████████████████████████████████▏                   | 190/279 [12:52<06:12,  4.19s/it]\u001b[A\n",
      "OllamaEmbeddings:  68%|██████████████████████████████████████████▍                   | 191/279 [12:55<05:49,  3.97s/it]\u001b[A\n",
      "OllamaEmbeddings:  69%|██████████████████████████████████████████▋                   | 192/279 [12:59<05:36,  3.86s/it]\u001b[A\n",
      "OllamaEmbeddings:  69%|██████████████████████████████████████████▉                   | 193/279 [13:04<05:58,  4.17s/it]\u001b[A\n",
      "OllamaEmbeddings:  70%|███████████████████████████████████████████                   | 194/279 [13:09<06:17,  4.44s/it]\u001b[A\n",
      "OllamaEmbeddings:  70%|███████████████████████████████████████████▎                  | 195/279 [13:13<06:17,  4.50s/it]\u001b[A\n",
      "OllamaEmbeddings:  70%|███████████████████████████████████████████▌                  | 196/279 [13:18<06:16,  4.54s/it]\u001b[A\n",
      "OllamaEmbeddings:  71%|███████████████████████████████████████████▊                  | 197/279 [13:23<06:26,  4.72s/it]\u001b[A\n",
      "OllamaEmbeddings:  71%|████████████████████████████████████████████                  | 198/279 [13:28<06:13,  4.61s/it]\u001b[A\n",
      "OllamaEmbeddings:  71%|████████████████████████████████████████████▏                 | 199/279 [13:32<06:10,  4.63s/it]\u001b[A\n",
      "OllamaEmbeddings:  72%|████████████████████████████████████████████▍                 | 200/279 [13:37<06:02,  4.59s/it]\u001b[A\n",
      "OllamaEmbeddings:  72%|████████████████████████████████████████████▋                 | 201/279 [13:41<05:56,  4.57s/it]\u001b[A\n",
      "OllamaEmbeddings:  72%|████████████████████████████████████████████▉                 | 202/279 [13:45<05:43,  4.47s/it]\u001b[A\n",
      "OllamaEmbeddings:  73%|█████████████████████████████████████████████                 | 203/279 [13:50<05:44,  4.54s/it]\u001b[A\n",
      "OllamaEmbeddings:  73%|█████████████████████████████████████████████▎                | 204/279 [13:54<05:24,  4.33s/it]\u001b[A\n",
      "OllamaEmbeddings:  73%|█████████████████████████████████████████████▌                | 205/279 [13:58<05:04,  4.12s/it]\u001b[A\n",
      "OllamaEmbeddings:  74%|█████████████████████████████████████████████▊                | 206/279 [14:01<04:47,  3.94s/it]\u001b[A\n",
      "OllamaEmbeddings:  74%|██████████████████████████████████████████████                | 207/279 [14:05<04:35,  3.83s/it]\u001b[A\n",
      "OllamaEmbeddings:  75%|██████████████████████████████████████████████▏               | 208/279 [14:08<04:24,  3.73s/it]\u001b[A\n",
      "OllamaEmbeddings:  75%|██████████████████████████████████████████████▍               | 209/279 [14:12<04:20,  3.73s/it]\u001b[A\n",
      "OllamaEmbeddings:  75%|██████████████████████████████████████████████▋               | 210/279 [14:15<03:57,  3.45s/it]\u001b[A\n",
      "OllamaEmbeddings:  76%|██████████████████████████████████████████████▉               | 211/279 [14:18<03:49,  3.37s/it]\u001b[A\n",
      "OllamaEmbeddings:  76%|███████████████████████████████████████████████               | 212/279 [14:21<03:43,  3.34s/it]\u001b[A\n",
      "OllamaEmbeddings:  76%|███████████████████████████████████████████████▎              | 213/279 [14:24<03:30,  3.19s/it]\u001b[A\n",
      "OllamaEmbeddings:  77%|███████████████████████████████████████████████▌              | 214/279 [14:27<03:22,  3.11s/it]\u001b[A\n",
      "OllamaEmbeddings:  77%|███████████████████████████████████████████████▊              | 215/279 [14:30<03:11,  3.00s/it]\u001b[A\n",
      "OllamaEmbeddings:  77%|████████████████████████████████████████████████              | 216/279 [14:33<03:18,  3.15s/it]\u001b[A\n",
      "OllamaEmbeddings:  78%|████████████████████████████████████████████████▏             | 217/279 [14:37<03:22,  3.27s/it]\u001b[A\n",
      "OllamaEmbeddings:  78%|████████████████████████████████████████████████▍             | 218/279 [14:40<03:25,  3.37s/it]\u001b[A\n",
      "OllamaEmbeddings:  78%|████████████████████████████████████████████████▋             | 219/279 [14:44<03:19,  3.32s/it]\u001b[A\n",
      "OllamaEmbeddings:  79%|████████████████████████████████████████████████▉             | 220/279 [14:47<03:13,  3.27s/it]\u001b[A\n",
      "OllamaEmbeddings:  79%|█████████████████████████████████████████████████             | 221/279 [14:50<03:06,  3.22s/it]\u001b[A\n",
      "OllamaEmbeddings:  80%|█████████████████████████████████████████████████▎            | 222/279 [14:53<03:00,  3.16s/it]\u001b[A\n",
      "OllamaEmbeddings:  80%|█████████████████████████████████████████████████▌            | 223/279 [14:56<02:59,  3.21s/it]\u001b[A\n",
      "OllamaEmbeddings:  80%|█████████████████████████████████████████████████▊            | 224/279 [14:59<02:52,  3.14s/it]\u001b[A\n",
      "OllamaEmbeddings:  81%|██████████████████████████████████████████████████            | 225/279 [15:02<02:52,  3.19s/it]\u001b[A\n",
      "OllamaEmbeddings:  81%|██████████████████████████████████████████████████▏           | 226/279 [15:05<02:46,  3.14s/it]\u001b[A\n",
      "OllamaEmbeddings:  81%|██████████████████████████████████████████████████▍           | 227/279 [15:09<02:47,  3.21s/it]\u001b[A\n",
      "OllamaEmbeddings:  82%|██████████████████████████████████████████████████▋           | 228/279 [15:12<02:38,  3.10s/it]\u001b[A\n",
      "OllamaEmbeddings:  82%|██████████████████████████████████████████████████▉           | 229/279 [15:14<02:28,  2.97s/it]\u001b[A\n",
      "OllamaEmbeddings:  82%|███████████████████████████████████████████████████           | 230/279 [15:17<02:21,  2.89s/it]\u001b[A\n",
      "OllamaEmbeddings:  83%|███████████████████████████████████████████████████▎          | 231/279 [15:20<02:15,  2.82s/it]\u001b[A\n",
      "OllamaEmbeddings:  83%|███████████████████████████████████████████████████▌          | 232/279 [15:22<02:10,  2.78s/it]\u001b[A\n",
      "OllamaEmbeddings:  84%|███████████████████████████████████████████████████▊          | 233/279 [15:25<02:06,  2.75s/it]\u001b[A\n",
      "OllamaEmbeddings:  84%|████████████████████████████████████████████████████          | 234/279 [15:28<02:01,  2.71s/it]\u001b[A\n",
      "OllamaEmbeddings:  84%|████████████████████████████████████████████████████▏         | 235/279 [15:30<01:58,  2.69s/it]\u001b[A\n",
      "OllamaEmbeddings:  85%|████████████████████████████████████████████████████▍         | 236/279 [15:33<01:54,  2.67s/it]\u001b[A\n",
      "OllamaEmbeddings:  85%|████████████████████████████████████████████████████▋         | 237/279 [15:36<01:51,  2.65s/it]\u001b[A\n",
      "OllamaEmbeddings:  85%|████████████████████████████████████████████████████▉         | 238/279 [15:38<01:48,  2.64s/it]\u001b[A\n",
      "OllamaEmbeddings:  86%|█████████████████████████████████████████████████████         | 239/279 [15:41<01:45,  2.63s/it]\u001b[A\n",
      "OllamaEmbeddings:  86%|█████████████████████████████████████████████████████▎        | 240/279 [15:43<01:43,  2.64s/it]\u001b[A\n",
      "OllamaEmbeddings:  86%|█████████████████████████████████████████████████████▌        | 241/279 [15:46<01:40,  2.64s/it]\u001b[A\n",
      "OllamaEmbeddings:  87%|█████████████████████████████████████████████████████▊        | 242/279 [15:49<01:38,  2.66s/it]\u001b[A\n",
      "OllamaEmbeddings:  87%|██████████████████████████████████████████████████████        | 243/279 [15:53<01:47,  2.98s/it]\u001b[A\n",
      "OllamaEmbeddings:  87%|██████████████████████████████████████████████████████▏       | 244/279 [15:56<01:50,  3.16s/it]\u001b[A\n",
      "OllamaEmbeddings:  88%|██████████████████████████████████████████████████████▍       | 245/279 [16:00<01:52,  3.32s/it]\u001b[A\n",
      "OllamaEmbeddings:  88%|██████████████████████████████████████████████████████▋       | 246/279 [16:03<01:50,  3.36s/it]\u001b[A\n",
      "OllamaEmbeddings:  89%|██████████████████████████████████████████████████████▉       | 247/279 [16:07<01:48,  3.38s/it]\u001b[A\n",
      "OllamaEmbeddings:  89%|███████████████████████████████████████████████████████       | 248/279 [16:09<01:37,  3.14s/it]\u001b[A\n",
      "OllamaEmbeddings:  89%|███████████████████████████████████████████████████████▎      | 249/279 [16:12<01:31,  3.04s/it]\u001b[A\n",
      "OllamaEmbeddings:  90%|███████████████████████████████████████████████████████▌      | 250/279 [16:15<01:25,  2.95s/it]\u001b[A\n",
      "OllamaEmbeddings:  90%|███████████████████████████████████████████████████████▊      | 251/279 [16:18<01:28,  3.17s/it]\u001b[A\n",
      "OllamaEmbeddings:  90%|████████████████████████████████████████████████████████      | 252/279 [16:22<01:30,  3.37s/it]\u001b[A\n",
      "OllamaEmbeddings:  91%|████████████████████████████████████████████████████████▏     | 253/279 [16:26<01:30,  3.47s/it]\u001b[A\n",
      "OllamaEmbeddings:  91%|████████████████████████████████████████████████████████▍     | 254/279 [16:29<01:20,  3.23s/it]\u001b[A\n",
      "OllamaEmbeddings:  91%|████████████████████████████████████████████████████████▋     | 255/279 [16:31<01:13,  3.08s/it]\u001b[A\n",
      "OllamaEmbeddings:  92%|████████████████████████████████████████████████████████▉     | 256/279 [16:34<01:07,  2.95s/it]\u001b[A\n",
      "OllamaEmbeddings:  92%|█████████████████████████████████████████████████████████     | 257/279 [16:37<01:03,  2.88s/it]\u001b[A\n",
      "OllamaEmbeddings:  92%|█████████████████████████████████████████████████████████▎    | 258/279 [16:39<00:59,  2.81s/it]\u001b[A\n",
      "OllamaEmbeddings:  93%|█████████████████████████████████████████████████████████▌    | 259/279 [16:43<01:03,  3.17s/it]\u001b[A\n",
      "OllamaEmbeddings:  93%|█████████████████████████████████████████████████████████▊    | 260/279 [16:49<01:14,  3.92s/it]\u001b[A\n",
      "OllamaEmbeddings:  94%|██████████████████████████████████████████████████████████    | 261/279 [16:53<01:12,  4.04s/it]\u001b[A\n",
      "OllamaEmbeddings:  94%|██████████████████████████████████████████████████████████▏   | 262/279 [16:58<01:10,  4.14s/it]\u001b[A\n",
      "OllamaEmbeddings:  94%|██████████████████████████████████████████████████████████▍   | 263/279 [17:01<01:03,  4.00s/it]\u001b[A\n",
      "OllamaEmbeddings:  95%|██████████████████████████████████████████████████████████▋   | 264/279 [17:04<00:54,  3.64s/it]\u001b[A\n",
      "OllamaEmbeddings:  95%|██████████████████████████████████████████████████████████▉   | 265/279 [17:07<00:46,  3.35s/it]\u001b[A\n",
      "OllamaEmbeddings:  95%|███████████████████████████████████████████████████████████   | 266/279 [17:10<00:40,  3.14s/it]\u001b[A\n",
      "OllamaEmbeddings:  96%|███████████████████████████████████████████████████████████▎  | 267/279 [17:12<00:35,  3.00s/it]\u001b[A\n",
      "OllamaEmbeddings:  96%|███████████████████████████████████████████████████████████▌  | 268/279 [17:15<00:32,  2.91s/it]\u001b[A\n",
      "OllamaEmbeddings:  96%|███████████████████████████████████████████████████████████▊  | 269/279 [17:18<00:28,  2.88s/it]\u001b[A\n",
      "OllamaEmbeddings:  97%|████████████████████████████████████████████████████████████  | 270/279 [17:20<00:25,  2.80s/it]\u001b[A\n",
      "OllamaEmbeddings:  97%|████████████████████████████████████████████████████████████▏ | 271/279 [17:23<00:23,  2.88s/it]\u001b[A\n",
      "OllamaEmbeddings:  97%|████████████████████████████████████████████████████████████▍ | 272/279 [17:26<00:20,  2.90s/it]\u001b[A\n",
      "OllamaEmbeddings:  98%|████████████████████████████████████████████████████████████▋ | 273/279 [17:30<00:17,  2.97s/it]\u001b[A\n",
      "OllamaEmbeddings:  98%|████████████████████████████████████████████████████████████▉ | 274/279 [17:34<00:17,  3.43s/it]\u001b[A\n",
      "OllamaEmbeddings:  99%|█████████████████████████████████████████████████████████████ | 275/279 [17:37<00:12,  3.20s/it]\u001b[A\n",
      "OllamaEmbeddings:  99%|█████████████████████████████████████████████████████████████▎| 276/279 [17:39<00:09,  3.03s/it]\u001b[A\n",
      "OllamaEmbeddings:  99%|█████████████████████████████████████████████████████████████▌| 277/279 [17:42<00:05,  2.90s/it]\u001b[A\n",
      "OllamaEmbeddings: 100%|█████████████████████████████████████████████████████████████▊| 278/279 [17:46<00:03,  3.13s/it]\u001b[A\n",
      "OllamaEmbeddings: 100%|██████████████████████████████████████████████████████████████| 279/279 [17:49<00:00,  3.83s/it]\u001b[A\n"
     ]
    }
   ],
   "source": [
    "vector_db = Chroma.from_texts(\n",
    "    texts=texts, \n",
    "    embedding=OllamaEmbeddings(model=\"nomic-embed-text\",show_progress=True),\n",
    "    collection_name=\"local-rag\"\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "id": "6e1f23d9-106a-4ae8-a089-f664ed2e6a8d",
   "metadata": {},
   "outputs": [],
   "source": [
    "local_model = \"llama2\"\n",
    "llm = ChatOllama(model=local_model)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "id": "303c4d48-da03-4c52-850a-1f6f52c4ff0a",
   "metadata": {},
   "outputs": [],
   "source": [
    "QUERY_PROMPT = PromptTemplate(\n",
    "    input_variables=[\"question\"],\n",
    "    template=\"\"\"You are an AI language model assistant. Your task is to generate five\n",
    "    different versions of the given user question to retrieve relevant documents from\n",
    "    a vector database. By generating multiple perspectives on the user question, your\n",
    "    goal is to help the user overcome some of the limitations of the distance-based\n",
    "    similarity search. Provide these alternative questions separated by newlines.\n",
    "    Original question: {question}\"\"\",\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "id": "0e11efef-131a-47c3-8702-2842d57955a5",
   "metadata": {},
   "outputs": [],
   "source": [
    "retriever = MultiQueryRetriever.from_llm(\n",
    "    vector_db.as_retriever(), \n",
    "    llm,\n",
    "    prompt=QUERY_PROMPT\n",
    ")\n",
    "\n",
    "# RAG prompt\n",
    "template = \"\"\"Answer the question based ONLY on the following context:\n",
    "{context}\n",
    "Question: {question}\n",
    "\"\"\"\n",
    "\n",
    "prompt = ChatPromptTemplate.from_template(template)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "id": "3e365621-0c5c-4e70-aa39-1598d7468265",
   "metadata": {},
   "outputs": [],
   "source": [
    "chain = (\n",
    "    {\"context\": retriever, \"question\": RunnablePassthrough()}\n",
    "    | prompt\n",
    "    | llm\n",
    "    | StrOutputParser()\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "id": "b587a878-bf06-4fde-9ef8-174656779a7b",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\n",
      "OllamaEmbeddings:   0%|                                                                          | 0/1 [00:00<?, ?it/s]\u001b[A\n",
      "OllamaEmbeddings: 100%|██████████████████████████████████████████████████████████████████| 1/1 [00:03<00:00,  3.36s/it]\u001b[A\n",
      "\n",
      "OllamaEmbeddings:   0%|                                                                          | 0/1 [00:00<?, ?it/s]\u001b[A\n",
      "OllamaEmbeddings: 100%|██████████████████████████████████████████████████████████████████| 1/1 [00:02<00:00,  2.13s/it]\u001b[A\n",
      "\n",
      "OllamaEmbeddings:   0%|                                                                          | 0/1 [00:00<?, ?it/s]\u001b[A\n",
      "OllamaEmbeddings: 100%|██████████████████████████████████████████████████████████████████| 1/1 [00:02<00:00,  2.18s/it]\u001b[A\n",
      "\n",
      "OllamaEmbeddings:   0%|                                                                          | 0/1 [00:00<?, ?it/s]\u001b[A\n",
      "OllamaEmbeddings: 100%|██████████████████████████████████████████████████████████████████| 1/1 [00:02<00:00,  2.16s/it]\u001b[A\n",
      "\n",
      "OllamaEmbeddings:   0%|                                                                          | 0/1 [00:00<?, ?it/s]\u001b[A\n",
      "OllamaEmbeddings: 100%|██████████████████████████████████████████████████████████████████| 1/1 [00:02<00:00,  2.40s/it]\u001b[A\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "\"Delta Lake is an open-source, distributed data storage system designed to handle large-scale data processing and analytics workloads. It is a follow-up project to Apache Spark's Structured Streaming, which provides real-time data processing capabilities. Delta Lake aims to address some of the challenges associated with handling large amounts of data in a distributed environment, such as data consistency, fault tolerance, and scalability.\\n\\nAt its core, Delta Lake is a data storage system that uses a log-structured merge tree (LSM) to organize and store data. It stores data in Parquet format, which provides efficient compression and data processing capabilities. Delta Lake also supports various data sources, including Apache Hive, Apache Cassandra, and Amazon S3.\\n\\nSome key features of Delta Lake include:\\n\\n1. Transactional storage: Delta Lake provides transactional storage, which means that it stores data in a way that ensures data consistency and accuracy. This is achieved through the use of a distributed transaction log that records all changes made to the data.\\n2. ACID compliance: Delta Lake is designed to be ACID-compliant, which means that it follows a set of rules to ensure that database transactions are processed reliably and securely.\\n3. Distributed architecture: Delta Lake is designed to scale horizontally, allowing it to handle large amounts of data and high levels of concurrency.\\n4. Fault tolerance: Delta Lake is designed to be fault-tolerant, meaning that it can continue to operate even in the event of hardware failures or other disruptions.\\n5. Data lineage and debugging: Delta Lake provides a detailed record of all changes made to the data, which can be used for data lineage and debugging purposes.\\n6. Time travel: Delta Lake allows users to go back in time to a previous state of the data, which can be useful for debugging or recreating a previous version of the data.\\n\\nOverall, Delta Lake is designed to provide a scalable, fault-tolerant, and ACID-compliant data storage system that can handle large amounts of data and support real-time data processing and analytics workloads.\""
      ]
     },
     "execution_count": 35,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "chain.invoke(\"What is delta lake?\")"
   ]
  }
 ],
 "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.7"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}