Demo: DeltaTable, DeltaLog And Snapshots¶

import org.apache.spark.sql.SparkSession
assert(spark.isInstanceOf[SparkSession])

val name = "users"
sql(s"DROP TABLE IF EXISTS $name")
sql(s"""
    | CREATE TABLE $name (id bigint, name string, city string, country string)
    | USING delta
    """.stripMargin)

import org.apache.spark.sql.catalyst.TableIdentifier
val tid = TableIdentifier(name)

import org.apache.spark.sql.delta.DeltaLog
val deltaLog = DeltaLog.forTable(spark, tid)

val snapshot = deltaLog.update()
assert(snapshot.version == 0)

val state = snapshot.state

scala> :type state
org.apache.spark.sql.Dataset[org.apache.spark.sql.delta.actions.SingleAction]

// Show the changes (actions)
scala> snapshot.state.show
+----+----+------+--------------------+--------+----------+
| txn| add|remove|            metaData|protocol|commitInfo|
+----+----+------+--------------------+--------+----------+
|null|null|  null|                null|  [1, 2]|      null|
|null|null|  null|[5156c9e3-9668-43...|    null|      null|
+----+----+------+--------------------+--------+----------+

import io.delta.tables.DeltaTable
val dt = DeltaTable.forName(name)

scala> dt.history.select('version, 'operation, 'operationParameters, 'operationMetrics).show(truncate = false)
+-------+-------------------+------+--------+------------+--------------------+----+--------+---------+-----------+--------------+-------------+----------------+------------+
|version|          timestamp|userId|userName|   operation| operationParameters| job|notebook|clusterId|readVersion|isolationLevel|isBlindAppend|operationMetrics|userMetadata|
+-------+-------------------+------+--------+------------+--------------------+----+--------+---------+-----------+--------------+-------------+----------------+------------+
|      0|2020-09-29 10:31:30|  null|    null|CREATE TABLE|[isManaged -> tru...|null|    null|     null|       null|          null|         true|              []|        null|
+-------+-------------------+------+--------+------------+--------------------+----+--------+---------+-----------+--------------+-------------+----------------+------------+

val users = dt.toDF

scala> users.show
+---+----+----+-------+
| id|name|city|country|
+---+----+----+-------+
+---+----+----+-------+

val newUsers = Seq(
  (0L, "Agata", "Warsaw", "Poland"),
  (1L, "Bartosz", "Paris", "France")
).toDF("id", "name", "city", "country")

scala> newUsers.show
+---+-------+------+-------+
| id|   name|  city|country|
+---+-------+------+-------+
|  0|  Agata|Warsaw| Poland|
|  1|Bartosz| Paris| France|
+---+-------+------+-------+

newUsers.write.format("delta").mode("append").saveAsTable(name)
assert(deltaLog.snapshot.version == 1)

scala> users.show
+---+-------+------+-------+
| id|   name|  city|country|
+---+-------+------+-------+
|  1|Bartosz| Paris| France|
|  0|  Agata|Warsaw| Poland|
+---+-------+------+-------+

scala> dt.history.select('version, 'operation, 'operationParameters, 'operationMetrics).show(truncate = false)
+-------+------------+------------------------------------------------------------------------+-----------------------------------------------------------+
|version|operation   |operationParameters                                                     |operationMetrics                                           |
+-------+------------+------------------------------------------------------------------------+-----------------------------------------------------------+
|1      |WRITE       |[mode -> Append, partitionBy -> []]                                     |[numFiles -> 2, numOutputBytes -> 2299, numOutputRows -> 2]|
|0      |CREATE TABLE|[isManaged -> true, description ->, partitionBy -> [], properties -> {}]|[]                                                         |
+-------+------------+------------------------------------------------------------------------+-----------------------------------------------------------+