Create/Write a Delta Table

eventsDF = spark.read.parquet(eventsPath)

1. Convert data to a Delta table using the schema provided by the DataFrame

Use save Save to path delta table

deltaPath = f"{
      delta_path}/delta-events"
eventsDF.write.format("delta").mode("overwrite").save(deltaPath)

2. create a Delta table in the metastore

Use saveAsTable Save to table

eventsDF.write.format("delta").mode("overwrite").saveAsTable("delta_events")

3. Partition save , Use partitionBy preservation .

from pyspark.sql.functions import col
stateEventsDF = eventsDF.withColumn("state", col("geo.state"))

stateEventsDF.write.format("delta").mode("overwrite").partitionBy("state").option("overwriteSchema", "true").save(deltaPath)

Read from your Delta table

df = spark.read.format("delta").load(deltaPath)
display(df)

Access previous versions of table using Time Travel

• Use describe history + Table name View table version information

describe history delta_user

You can see that the table has 4 A version

• Use VERSION AS OF/versionAsOf Get historical version data

• Use TIMESTAMP AS OF/timestampAsOf Get historical version data

df = spark.read.format("delta").option("versionAsOf", 0).load(deltaPath)
df = spark.read.format("delta").option("timestampAsOf", "2021-10-23T11:07:57.000+0000").load(deltaPath)

SELECT * FROM delta_events TIMESTAMP AS OF '2021-10-23T11:07:57.000+0000'
SELECT * FROM delta.`/delta_events` VERSION AS OF 0

Vacuum

we can clean up our directory using VACUUM. Vacuum accepts a retention period in hours as an input.

By default, to prevent accidentally vacuuming recent commits, Delta Lake will not let users vacuum a period under 7 days or 168 hours. Once

vacuumed, you cannot return to a prior commit through time travel, only your most recent Delta Table will be saved.

from delta.tables import *

spark.conf.set("spark.databricks.delta.retentionDurationCheck.enabled", "false")
deltaTable = DeltaTable.forPath(spark, deltaPath)
deltaTable.vacuum(0)