what does the file Snapshot.scala in databricks?

0
votes

I am running some streaming query jobs on a databricks cluster, and when i look at the cluster/job logs, I see a lot of

first at Snapshot.scala:1

and

withNewExecutionId at TransactionalWriteEdge.scala:130

A quick search yielded this scala script https://github.com/delta-io/delta/blob/master/src/main/scala/org/apache/spark/sql/delta/Snapshot.scala Any one can explain what this do in laymans term?

1
scalaapache-sparkdatabricks

1 Answers

0
votes

Internally this class manages the replay of actions stored in checkpoint or delta file

Generally, this "snapshotting" relies on delta encoding and indirectly allows snaphot isolation as well.

Practically delta-encoding remembers every side-effectful operation like INSERT DELETE UPDATE that you did since the last checkpoint. In case of delta lake it would be SingleAction (source): AddFile (insert) RemoveFile (delete). Conceptually this approach is close to event-sourcing - without it you'd have to literally store/broadcast whole state (database or directory) on every update. It also employed by many classic ACID databases with replication.

Overall it gives you:

  • ability to continuously replicate file-system/directory/database state (see SnapshotManagement.update). Basically that's why you see a lot of first at Snapshot.scala:1 - it's called in order to catch up with the log every time you start transaction, see DeltaLog.startTransaction. I couldn't find TransactionalWriteEdge sources, but I guess it's called around the same time.
  • ability to restore state by replaying every action since the last snapshot.
  • ability to isolate (and store) transactions by keeping their snapshots apart until commit (every SingleAction has txn in order to isolate). Delta-lake uses optimistic locking for that: transaction commits will fail if their logs are not mergeable, while readers don't see uncommitted actions.

P.S. You can see that the log is accessed in line val deltaData = load(files) and actions are stacked on top of previousSnapshot (val checkpointData = previousSnapshot.getOrElse(emptyActions); val allActions = checkpointData.union(deltaData))