Parquet file
• Compression
• Columnar Storage
• Language Agnostics
• Open Source
• Complex Data Types
Lakehouse
Data Lake
Lakehouse
Delta Lake: it is a storage layer on the top of Data Lake/ Open source.
• Stores data in parquet file format that allows various features
• ACID Properties
• Columnar Storage
• Time Interval : version and timestamp of each parquet files
• Data Recovery
• through time interval
• Support for batch and streaming data.
How to Create Delta Table
• Download [Link]
• Open Notebook
• Add lakehouse
• Upload [Link] files in lakehouse explorer
• Load the [Link] to dataframe using # Load a file into a dataframe
df = [Link]('Files/[Link]', format='csv', header=True)
# Save the dataframe as a delta table
[Link]("delta").saveAsTable("Sales")
Types of Tables
• Managed Table
• External Table
• Managed Table is created by spark context so this managed delta table
• IF you specify the file path, it become external table
How to create External Table
df=[Link]('Files/[Link]',format='csv',header=true)
[Link]("delta").saveAsTable("SalesExternal",path="Files/myexternaltabl")
• This will create table and files in the lakehouse that will be external and
will persist.
• You can give fully qualified path abfs
[Link]("delta").saveAsTable("myexternaltable", path="abfss://my_store_url..../myexternaltable")
DeltaTableBuilder API
You can use python DeltaTableBuilder API to create and manage delta tables
from [Link] import *
[Link](spark) \
.tableName("products") \
.addColumn("Productid", "INT") \
.addColumn("ProductName", "STRING") \
.addColumn("Category", "STRING") \
.addColumn("Price", "FLOAT") \
.execute()
Spark SQL
You can use Spark SQL to create Managed and External Table
%%sql
CREATE TABLE salesorders
(
Orderid INT NOT NULL,
OrderDate TIMESTAMP NOT NULL,
CustomerName STRING,
SalesTotal FLOAT NOT NULL
)
USING DELTA
Delta Table with Spark SQL External
%%sql
CREATE TABLE Products(ProductID INT, Product Varchar(100), Category
varchar(100), Price float)
USING DELTA
LOCATION 'Files/mydata'
Save Dataframe to Delta Table
delta_path = "Files/mydatatable"
[Link]("delta").save(delta_path)
new_df.[Link]("delta").mode("overwrite").save(delta_path)
new_rows_df.[Link]("delta").mode("append").save(delta_path)
Run SQL Commands
[Link]("INSERT INTO Products(ProductID,Product,Category,Price)
Values(1,'Iphone','Phone',55444), (2,Macbook Pro’,Laptop’,155044)
, (3,Samsung Galaxy','Phone’,6000)
• %%sql
• UPDATE products
• SET Price = 2565677 WHERE ProductId = 1;
Run SQL Commands
[Link]("INSERT INTO Products(ProductID,Product,Category,Price)
Values(1,'Iphone','Phone',55444), (2,Macbook Pro’,Laptop’,155044)
, (3,Samsung Galaxy','Phone’,6000)
%%sql
UPDATE products
SET Price = 2565677 WHERE ProductId = 1;
Update using Delta AP
from [Link] import *
from [Link] import *
delta_path = "Files/[Link]"
deltaTable = [Link](spark, delta_path)
[Link](
condition = "Category == TV",
set = { "Price": "Price * 0.9" })
Time Travel
Used to fetch changes made on table over the period of time
%%sql
DESCRIBE HISTORY products
Using delta_log
Time Travel for External Tables
Used to fetch changes made on table over the period of time, specify file
path
%%sql
DESCRIBE HISTORY 'Files/mytable'
Retrieve Data from Specific Version
Retrieve data from a specific version of the data by reading the delta
file location into a dataframe
df = [Link]("delta").option("versionAsOf", 0).load(delta_path)
Using Timestamp
df = [Link]("delta").option("timestampAsOf", '2022-01-
01').load(delta_path)
Streaming
• delta tables with streaming data
• Spark Structured Streaming APIS
• Source can be : IoT, Network Ports, Real time message broking services
such as Kafka, Azure Event Hub
How it work In real world?
Our Use case
Source Sink
Delta Table
{"device":"Sensor1","statu
s":"ok"}
Create Stream
from [Link] import *
from [Link] import *
# Create a stream that reads JSON data from a folder
inputPath = 'Files/streamingdata/'
jsonSchema = StructType([
StructField("device", StringType(), False),
StructField("status", StringType(), False)
])
stream_df = [Link](jsonSchema).option("maxFilesPerTrigger", 1).json(inputPath)
# Write the stream to a delta table
table_path = 'Files/delta/devicetable'
checkpoint_path = 'Files/delta/checkpoint'
delta_stream = stream_df.[Link]("delta").option("checkpointLocation", checkpoint_path).start(table_path)
Create Table
Create table fromdata
visualize Sink
%%sql
CREATE TABLE DeviceTable
USING DELTA
LOCATION 'Files/delta/devicetable';
SELECT device, status
FROM DeviceTable;
Stop
Stop Strema
Stream
• delta_stream.stop()