Delta Lake 事务日志实现的源码剖析

笔者在之前的文章《实战深入理解 Delta Lake 事务日志》中带大家把 Delta Lake 的事务日志大致操作了一遍，并进行了具体的分析。

有了之前的基础，笔者将在本篇文章中继续和大家一起深入研究 Delta Lake 事务日志的源码实现，学习 Delta Lake 开源项目的工程经验。

环境信息

Delta Lake Github：

https://github.com/delta-io/delta

笔者选取的版本为最新发布版本 v0.4.0，源码下载地址为：

https://github.com/delta-io/delta/releases/tag/v0.4.0

看一下 Delta Lake 项目的目录结构：  

大部分代码实现都在 org.apache.spark.sql.delta 包下面。代码整体层次还是很清晰的，Scala 编程语言实现。

Delta Lake 事务日志源码分析

读者最好先大体看一下代码结构，点点看。

有没有发现什么？有的读者，可能发现了什么，不知道从哪里入手。

但是要不了多久，聪明的读者会发现 DeltaLog 这个类，打开看看。

org.apache.spark.sql.delta.DeltaLog

1. /**

2. * Used to query the current state of the log as well as modify it by adding

3. * new atomic collections of actions.

4. *

5. * Internally, this class implements an optimistic concurrency control

6. * algorithm to handle multiple readers or writers. Any single read

7. * is guaranteed to see a consistent snapshot of the table.

8. */

DeltaLog 类的注释中有一句话很重要：

Internally, this class implements an optimistic concurrency control algorithm to handle multiple readers or writers. Any single read is guaranteed to see a consistent snapshot of the table.

大致意思为：

在内部，DeltaLog 类实现了一个乐观并发控制算法来处理并发读取或写入操作。任何一次读操作都保证看到表的一致性快照。

为了更方便的分析，我们直接看一下事务开始的代码：

1. /* ------------------ *

2. | Delta Management |

3. * ------------------ */

4. 
   

5. /**

6. * Returns a new [[OptimisticTransaction]] that can be used to read the current state of the

7. * log and then commit updates. The reads and updates will be checked for logical conflicts

8. * with any concurrent writes to the log.

9. *

10. * Note that all reads in a transaction must go through the returned transaction object, and not

11. * directly to the [[DeltaLog]] otherwise they will not be checked for conflicts.

12. */

13. def startTransaction(): OptimisticTransaction = {

14. update()

15. new OptimisticTransaction(this)

16. }

其实这里注释说的不是很清楚，不着急，我们接着分析。但是这里出现的 OptimisticTransaction 类是事务日志的关键类，对于事务日志的持久化都需要通过这个类，这也正是上面所提到的乐观事务，下面我们将具体分析该类。

OptimisticTransaction 类，直接看名字的意思很明确，乐观事务。该类维护了一个 case class，即 CommitStats。CommitStats 记录了一个成功的事务提交的 metrics，如下：

1. case class CommitStats(

2. /** The version read by the txn when it starts. */

3. startVersion: Long,

4. /** The version committed by the txn. */

5. commitVersion: Long,

6. /** The version read by the txn right after it commits. It usually equals to commitVersion,

7. * but can be larger than commitVersion when there are concurrent commits. */

8. readVersion: Long,

9. txnDurationMs: Long,

10. commitDurationMs: Long,

11. numAdd: Int,

12. numRemove: Int,

13. bytesNew: Long,

14. /** The number of files in the table as of version `readVersion`. */

15. numFilesTotal: Long,

16. /** The table size in bytes as of version `readVersion`. */

17. sizeInBytesTotal: Long,

18. /** The protocol as of version `readVersion`. */

19. protocol: Protocol,

20. info: CommitInfo,

21. newMetadata: Option[Metadata],

22. numAbsolutePathsInAdd: Int,

23. numDistinctPartitionsInAdd: Int,

24. isolationLevel: String)

OptimisticTransaction 类定义如下实现内容，包含一些笔者额外标记的注释，具体分析请继续看后文：

1. ...

2. trait OptimisticTransactionImpl extends TransactionalWrite {

3. ...

4. // commit 方法，参数见后文说明

5. @throws(classOf[ConcurrentModificationException])

6. def commit(actions: Seq[Action], op: DeltaOperations.Operation): Long = recordDeltaOperation(

7. deltaLog,

8. "delta.commit") {

9. val version = try {

10. // 事务日志提交前的准备工作

11. // Try to commit at the next version.

12. var finalActions = prepareCommit(actions, op)

13. // 如果本次更新要删除之前文件，则 isBlindAppend 为 false，否则为 true

14. val isBlindAppend = {

15. val onlyAddFiles =

16. finalActions.collect { case f: FileAction => f }.forall(_.isInstanceOf[AddFile])

17. onlyAddFiles && !dependsOnFiles

18. }

19. // 如果 commitInfo.enabled 参数设置为 true，则需要把 commitInfo 记录到事务日志里面

20. if (spark.sessionState.conf.getConf(DeltaSQLConf.DELTA_COMMIT_INFO_ENABLED)) {

21. commitInfo = CommitInfo(

22. clock.getTimeMillis(),

23. op.name,

24. op.jsonEncodedValues,

25. Map.empty,

26. Some(readVersion).filter(_ >= 0),

27. None,

28. Some(isBlindAppend))

29. finalActions = commitInfo +: finalActions

30. }

31. // 开始写事务日志，如果检测到任何冲突，会尝试解决逻辑冲突并使用新版本提交

32. val commitVersion = doCommit(snapshot.version + 1, finalActions, 0)

33. logInfo(s"Committed delta #$commitVersion to ${deltaLog.logPath}")

34. // 对事务日志执行 checkpoint 操作

35. postCommit(commitVersion, finalActions)

36. commitVersion

37. } catch {

38. case e: DeltaConcurrentModificationException =>

39. recordDeltaEvent(deltaLog, "delta.commit.conflict." + e.conflictType)

40. throw e

41. case NonFatal(e) =>

42. recordDeltaEvent(

43. deltaLog, "delta.commit.failure", data = Map("exception" -> Utils.exceptionString(e)))

44. throw e

45. }

46. version

47. }

48. ...

为了方便分析和以后查看，我贴了该 commit 方法的全部实现，请读者忍受一下。这个方法非常重要，包含大部分事务日志实现的代码。

commit 方法的参数

commit 方法定义：

1. def commit(actions: Seq[Action], op: DeltaOperations.Operation): Long = recordDeltaOperation {

2. ...

3. }

• actions: Seq[Action]

  记录事务记录（SetTransaction）

  表更新操作产生的新文件（AddFile）

  删除文件（RemoveFile）

  元数据（metaData）

  更新操作首次初始化protocol（Protocol）

  提交信息（CommitInfo）

• op: DeltaOperations.Operation

  Delta 操作类型，包括Write、StreamingUpdate、Delete、Truncate、Update等一系列操作类型，具体请查看 DeltaOperations.scala 。

commit 方法的三部曲

整体看完 commit 方法后，主要分为三部分内容：

• 1. prepareCommit
  

• 2. doCommit
  

• 3. postCommit
  

1. prepareCommit 方法

1. protected def prepareCommit(

2. actions: Seq[Action],

3. op: DeltaOperations.Operation): Seq[Action] = {

4. // 事务是否已经提交，增加断言

5. assert(!committed, "Transaction already committed.")

6. 
   

7. // 1. 如果更新了表的 Metadata 信息，那么需要将其写入到事务日志里面

8. // If the metadata has changed, add that to the set of actions

9. var finalActions = newMetadata.toSeq ++ actions

10. val metadataChanges = finalActions.collect { case m: Metadata => m }

11. assert(

12. metadataChanges.length <= 1,

13. "Cannot change the metadata more than once in a transaction.")

14. metadataChanges.foreach(m => verifyNewMetadata(m))

15. 
    

16. // 2. 首次提交事务日志，那么会确保 _delta_log 目录要存在，然后检查 finalActions 里面是否有 Protocol，没有的话需要初始化 protocol 版本

17. if (snapshot.version == -1) {

18. deltaLog.ensureLogDirectoryExist()

19. if (!finalActions.exists(_.isInstanceOf[Protocol])) {

20. finalActions = Protocol() +: finalActions

21. }

22. }

23. 
    

24. finalActions = finalActions.map {

25. // 3. 当第一次提交，并且有 Metadata，那么会将 Delta Lake 的全局配置信息加入到 Metadata 里面

26. case m: Metadata if snapshot.version == -1 =>

27. val updatedConf = DeltaConfigs.mergeGlobalConfigs(

28. spark.sessionState.conf, m.configuration, Protocol())

29. m.copy(configuration = updatedConf)

30. case other => other

31. }

32. 
    

33. deltaLog.protocolWrite(

34. snapshot.protocol,

35. logUpgradeMessage = !actions.headOption.exists(_.isInstanceOf[Protocol]))

36. 
    

37. // 4. 在检查是否需要删除文件时，我们要确保这不是一个 appendOnly 表。

38. val removes = actions.collect { case r: RemoveFile => r }

39. if (removes.exists(_.dataChange)) deltaLog.assertRemovable()

40. 
    

41. finalActions

42. }

prepareCommit 里面的重要操作，根据代码的注释标记了1、2、3和4，具体为：

• 1. 由于 Delta Lake 表允许对已经存在的表模式进行修改，比如添加新列或者覆盖原有表的模式等，需要将新的 Metadata 写入到事务日志里面。Metadata 里面存储了表的 schema、分区列、表的配置、表的创建时间等信息，如下：

1. case class Metadata(

2. id: String = java.util.UUID.randomUUID().toString,

3. name: String = null,

4. description: String = null,

5. format: Format = Format(),

6. schemaString: String = null,

7. partitionColumns: Seq[String] = Nil,

8. configuration: Map[String, String] = Map.empty,

9. @JsonDeserialize(contentAs = classOf[java.lang.Long])

10. createdTime: Option[Long] = Some(System.currentTimeMillis())

• 2. 如果是首次提交事务日志，那么先检查表的 _delta_log 目录是否存在，不存在则创建。然后检查是否设置了 protocol 的版本，如果没有设置，则使用默认的协议版本，默认的协议版本中 Action.readerVersion = 1，Action.writerVersion = 2。
  

• 3. 如果是第一次提交，并且存在 Metadata ，那么会将 Delta Lake 的配置信息加入到 Metadata 里面。Delta Lake 表的配置信息主要是在 org.apache.spark.sql.delta.sources.DeltaSQLConf 类里面定义的，比如可以在创建 Delta Lake 表的时候指定多久做一次 Checkpoint。
  

• 4. 可以通过 spark.databricks.delta.properties.defaults.appendOnly 参数将表设置为仅允许追加，所以如果当 actions 里面存在 RemoveFile，那么我们需要判断表是否允许删除。
  

prepareCommit 方法的返回值为 finalActions，这些信息就是需要写入到事务日志里面的数据。

1. var finalActions = prepareCommit(actions, op)

2. 
   

3. val isBlindAppend = {

4. val onlyAddFiles =

5. finalActions.collect { case f: FileAction => f }.forall(_.isInstanceOf[AddFile])

6. onlyAddFiles && !dependsOnFiles

7. }

紧接着会判断这次事务变更是否需要删除之前的文件，如果是，那么 isBlindAppend 为 false，否则为 true。

1. if (spark.sessionState.conf.getConf(DeltaSQLConf.DELTA_COMMIT_INFO_ENABLED)) {

2. commitInfo = CommitInfo(

3. clock.getTimeMillis(),

4. op.name,

5. op.jsonEncodedValues,

6. Map.empty,

7. Some(readVersion).filter(_ >= 0),

8. None,

9. Some(isBlindAppend))

10. finalActions = commitInfo +: finalActions

11. }

当 commitInfo.enabled 参数设置为 true（默认值），那么还需要将 commitInfo 写入到事务日志文件里面。CommitInfo 里面包含了操作时间、操作的类型（Write、Update等）等重要信息。

接下来开始调用 doCommit 方法。

2. doCommit 方法

doCommit 方法传入两个参数：

1. val commitVersion = doCommit(snapshot.version + 1, finalActions, 0)

doCommit 方法的第一个参数传递是 snapshot.version + 1。snapshot.version 其实就是事务日志中最新的版本，我们再来查看一下 Delta Lake 表的目录信息： 

如果snapshot.version 的值为1，那么这次 doCommit 的更新版本为 2。

doCommit 方法具体内容如下：

1. private def doCommit(

2. attemptVersion: Long,

3. actions: Seq[Action],

4. attemptNumber: Int): Long = deltaLog.lockInterruptibly {

5. try {

6. logDebug(s"Attempting to commit version $attemptVersion with ${actions.size} actions")

7. // 1. 正式写事务日志的操作

8. deltaLog.store.write(

9. deltaFile(deltaLog.logPath, attemptVersion),

10. actions.map(_.json).toIterator)

11. val commitTime = System.nanoTime()

12. // 2. 由于发生了数据更新，所以更新内存中事务日志的最新快照，并做相关判断

13. val postCommitSnapshot = deltaLog.update()

14. if (postCommitSnapshot.version < attemptVersion) {

15. throw new IllegalStateException(

16. s"The committed version is $attemptVersion " +

17. s"but the current version is ${postCommitSnapshot.version}.")

18. }

19. 
    

20. // 3. 发送一些统计信息

21. var numAbsolutePaths = 0

22. var pathHolder: Path = null

23. val distinctPartitions = new mutable.HashSet[Map[String, String]]

24. val adds = actions.collect {

25. case a: AddFile =>

26. pathHolder = new Path(new URI(a.path))

27. if (pathHolder.isAbsolute) numAbsolutePaths += 1

28. distinctPartitions += a.partitionValues

29. a

30. }

31. val stats = CommitStats(

32. startVersion = snapshot.version,

33. commitVersion = attemptVersion,

34. readVersion = postCommitSnapshot.version,

35. txnDurationMs = NANOSECONDS.toMillis(commitTime - txnStartNano),

36. commitDurationMs = NANOSECONDS.toMillis(commitTime - commitStartNano),

37. numAdd = adds.size,

38. numRemove = actions.collect { case r: RemoveFile => r }.size,

39. bytesNew = adds.filter(_.dataChange).map(_.size).sum,

40. numFilesTotal = postCommitSnapshot.numOfFiles,

41. sizeInBytesTotal = postCommitSnapshot.sizeInBytes,

42. protocol = postCommitSnapshot.protocol,

43. info = Option(commitInfo).map(_.copy(readVersion = None, isolationLevel = None)).orNull,

44. newMetadata = newMetadata,

45. numAbsolutePaths,

46. numDistinctPartitionsInAdd = distinctPartitions.size,

47. isolationLevel = null)

48. recordDeltaEvent(deltaLog, "delta.commit.stats", data = stats)

49. 
    

50. attemptVersion

51. } catch {

52. case e: java.nio.file.FileAlreadyExistsException =>

53. checkAndRetry(attemptVersion, actions, attemptNumber)

54. }

55. }

根据注释标记的数字顺序介绍：

• 1. 正式写事务日志的操作，其中 store 是通过 spark.delta.logStore.class 参数指定的，目前支持 HDFS、S3、Local 等存储介质，默认是 HDFS。具体的写事务操作的过程，接下来介绍。
  

• 2. 持久化事务日志之后，更新内存中的事务日志最新的快照，然后做相关的合法性校验。
  

• 3. 发送一些统计信息。
  

我们针对 deltaLog 写事务日志操作专门进行解说：

1. deltaLog.store.write(

2. deltaFile(deltaLog.logPath, attemptVersion),

3. actions.map(_.json).toIterator

4. )

write 方法传入两个参数：

• HDFS路径，deltaFile 方法返回值

1. /** Returns the path for a given delta file. */

2. def deltaFile(path: Path, version: Long): Path = new Path(path, f"$version%020d.json")

• actions，doCommit 方法传入的参数 finalActions

write 方法的实现支持好几种存储，比如HDFS、S3、Azure等，这里以大数据平台常用的 HDFS 分布式存储系统来分析。 

HDFSLogStore 类实现 LogStore 接口，查看 write 方法的实现：

1. def write(path: Path, actions: Iterator[String], overwrite: Boolean = false): Unit = {

2. val isLocalFs = path.getFileSystem(getActiveHadoopConf).isInstanceOf[RawLocalFileSystem]

3. if (isLocalFs) {

4. synchronized {

5. writeInternal(path, actions, overwrite)

6. }

7. } else {

8. writeInternal(path, actions, overwrite)

9. }

10. }

其实 write 调用的核心方法为 writeInternal，如下：

1. private def writeInternal(path: Path, actions: Iterator[String], overwrite: Boolean): Unit = {

2. // 1. 获取 HDFS 的 FileContext 用于后面写事务日志

3. val fc = getFileContext(path)

4. 
   

5. // 2. 如果需要写的事务日志已经存在那么就需要抛出异常，后面再重试

6. if (!overwrite && fc.util.exists(path)) {

7. // This is needed for the tests to throw error with local file system

8. throw new FileAlreadyExistsException(path.toString)

9. }

10. 
    

11. // 3. 事务日志先写到临时文件

12. val tempPath = createTempPath(path)

13. var streamClosed = false // This flag is to avoid double close

14. var renameDone = false // This flag is to save the delete operation in most of cases.

15. val stream = fc.create(

16. tempPath, EnumSet.of(CREATE), CreateOpts.checksumParam(ChecksumOpt.createDisabled()))

17. 
    

18. try {

19. // 4. 将本次修改产生的 actions 写入到临时事务日志里

20. actions.map(_ + "\n").map(_.getBytes(UTF_8)).foreach(stream.write)

21. stream.close()

22. streamClosed = true

23. try {

24. val renameOpt = if (overwrite) Options.Rename.OVERWRITE else Options.Rename.NONE

25. // 5. 将临时的事务日志移到正式的事务日志里面，如果移动失败则抛出异常，后面再重试

26. fc.rename(tempPath, path, renameOpt)

27. renameDone = true

28. // TODO: this is a workaround of HADOOP-16255 - remove this when HADOOP-16255 is resolved

29. tryRemoveCrcFile(fc, tempPath)

30. } catch {

31. case e: org.apache.hadoop.fs.FileAlreadyExistsException =>

32. throw new FileAlreadyExistsException(path.toString)

33. }

34. } finally {

35. if (!streamClosed) {

36. stream.close()

37. }

38. // 删除临时事务日志

39. if (!renameDone) {

40. fc.delete(tempPath, false)

41. }

42. }

43. }

writeInternal 方法的实现过程，就是对 HDFS 进行写文件操作，结合上面数字标记的顺序，具体说明如下：

1. 获取 HDFS 的 FileContext 用于写事务日志

2. 如果需要写的事务日志已经存在，那么就需要抛出异常，然后再重试

3. 写事务日志的时候是先写到表 _delta_lake 目录下的临时文件里面

4. 将本次更新操作的事务记录写到临时文件里

5. 写完事务日志之后，需要将临时事务日志最后移动动正式的日志文件里面。这里需要注意，在写事务日志文件的过程中同样存在多个用户修改表，拿 00000000000000000004.json 这个文件举例，很可能已经被别的修改占用了，这时候也需要抛出 FileAlreadyExistsException 异常，以便后面重试

到此，Delta Lake 的事务日志写操作就完成了。这里需要注意的是，doCommit 有可能会失败，抛出 FileAlreadyExistsException 异常。Delta Lake 在实现 doCommit 方法时捕获了这个异常，并在异常捕获里面调用 checkAndRetry(attemptVersion, actions, attemptNumber) 方法进行重试操作：

1. } catch {

2. case e: java.nio.file.FileAlreadyExistsException =>

3. checkAndRetry(attemptVersion, actions, attemptNumber)

4. }

checkAndRetry 方法非常简单，这里就不细说了，只是需要注意，重试的版本是刚刚更新内存中事务日志快照的版本加上1：

1. // 因为上次更新事务日志发生冲突，所以需要再一次读取磁盘中持久化的事务日志，并更新内存中事务日志快照

2. deltaLog.update()

3. // 重试的版本是刚刚更新内存中事务日志快照的 version + 1

4. val nextAttempt = deltaLog.snapshot.version + 1

checkAndRetry 方法只有在事务日志写冲突的时候才会出现，主要目的是重写当前的事务日志。

当事务日志成功持久化到磁盘之后，最后再执行 postCommit 操作。

3. postCommit 方法

1. protected def postCommit(commitVersion: Long, commitActions: Seq[Action]): Unit = {

2. committed = true

3. if (commitVersion != 0 && commitVersion % deltaLog.checkpointInterval == 0) {

4. try {

5. deltaLog.checkpoint()

6. } catch {

7. case e: IllegalStateException =>

8. logWarning("Failed to checkpoint table state.", e)

9. }

10. }

11. }

postCommit 的实现相对来说是最简单的，功能就是判断需不需要对事务日志做一次 checkpoint 操作。deltaLog.checkpointInterval 的值是通过 spark.databricks.delta.properties.defaults.checkpointInterval 参数设置的，默认每写10次事务日志做一次 checkpoint。

checkpoint 的其实就是将内存中事务日志的最新快照持久化到磁盘里面，如下： 

/delta/mydelta.db/user_info/_delta_log/00000000000000000010.checkpoint.parquet 文件就是对事务日志进行 checkpoint 的文件，里面汇总了 00000000000000000000.json - 00000000000000000010.json 之间的所有事务操作记录。那么下一次如果再构建事务日志的快照时，只需要从 00000000000000000010.checkpoint.parquet 文件以及往后更新的文件开始构造，而无需再读取 00000000000000000000.json 到 00000000000000000010.json 之间的事务操作。

另外，我们还可以从 HDFS 路径看出，checkpoint 之后还会生成一个 _last_checkpoint 文件，里面记录了最后一次 checkpoint 的版本，checkpoint 文件里面的 Action 条数，如下：

1. {"version":10,"size":13}

到此，笔者已经带大家完成了对  Delta Lake 事务日志的源码实现的研究，希望大家对 Delta Lake 的认识更深一层。